// Copyright 2023 The OPA Authors. All rights reserved. // Use of this source code is governed by an Apache2 // license that can be found in the LICENSE file. // Package pathwatcher provides helper functions for creating file and directory watchers package pathwatcher import ( "context" "os" "path/filepath" "github.com/fsnotify/fsnotify" initload "github.com/open-policy-agent/opa/internal/runtime/init" "github.com/open-policy-agent/opa/v1/ast" "github.com/open-policy-agent/opa/v1/loader" "github.com/open-policy-agent/opa/v1/storage" "github.com/open-policy-agent/opa/v1/util" ) // CreatePathWatcher creates watchers to monitor for path changes func CreatePathWatcher(rootPaths []string) (*fsnotify.Watcher, error) { watchPaths, err := getWatchPaths(rootPaths) if err != nil { return nil, err } watcher, err := fsnotify.NewWatcher() if err != nil { return nil, err } for _, path := range watchPaths { if err := watcher.Add(path); err != nil { return nil, err } } return watcher, nil } // ProcessWatcherUpdate handles an occurrence of a watcher event func ProcessWatcherUpdate(ctx context.Context, paths []string, removed string, store storage.Store, filter loader.Filter, asBundle bool, bundleLazyLoadingMode bool, f func(context.Context, storage.Transaction, *initload.LoadPathsResult) error) error { return ProcessWatcherUpdateForRegoVersion(ctx, ast.ParserOptions{RegoVersion: ast.DefaultRegoVersion, ProcessAnnotation: true}, paths, removed, store, filter, asBundle, bundleLazyLoadingMode, f) } func ProcessWatcherUpdateForRegoVersion(ctx context.Context, popts ast.ParserOptions, paths []string, removed string, store storage.Store, filter loader.Filter, asBundle bool, bundleLazyLoadingMode bool, f func(context.Context, storage.Transaction, *initload.LoadPathsResult) error) error { loaded, err := initload.LoadPathsForRegoVersion(popts, paths, filter, asBundle, nil, true, bundleLazyLoadingMode, false, nil) if err != nil { return err } removed = loader.CleanPath(removed) return storage.Txn(ctx, store, storage.WriteParams, func(txn storage.Transaction) error { if !asBundle { ids, err := store.ListPolicies(ctx, txn) if err != nil { return err } for _, id := range ids { if id == removed { if err := store.DeletePolicy(ctx, txn, id); err != nil { return err } } else if _, exists := loaded.Files.Modules[id]; !exists { // This branch get hit in two cases. // 1. Another piece of code has access to the store and inserts // a policy out-of-band. // 2. In between FS notification and loader.Filtered() call above, a // policy is removed from disk. bs, err := store.GetPolicy(ctx, txn, id) if err != nil { return err } module, err := ast.ParseModuleWithOpts(id, string(bs), popts) if err != nil { return err } loaded.Files.Modules[id] = &loader.RegoFile{ Name: id, Raw: bs, Parsed: module, } } } } return f(ctx, txn, loaded) }) } func getWatchPaths(rootPaths []string) ([]string, error) { paths := []string{} for _, path := range rootPaths { _, path = loader.SplitPrefix(path) result, err := loader.Paths(path, true) if err != nil { return nil, err } unique := map[string]struct{}{} for _, r := range result { fi, err := os.Lstat(r) if err != nil { return nil, err } if fi.IsDir() { unique[r] = struct{}{} } else { dir := filepath.Dir(r) unique[dir] = struct{}{} } } paths = append(paths, util.KeysSorted(unique)...) } return paths, nil }