Rate limit hunt dispatcher housekeeping thread

scudette · scudette · commit 231de332c01a · 2026-02-23T10:08:22.000+10:00
This ensures that this IO intensive operation does not impact on
server performance.

Also:
* Allow query to take a limit callback for early exit.
* Redesign the file upload GUI in the upload artifact pack screen.
* Fix watch_ebpf() to create a default policy if only events were
  passed - this is needed for backwards compatibility.
* The template() VQL function now accepts a `html` parameter which
  allows using go's html/template instead. This should be safe for
  generating HTML
diff --git a/actions/proto/vql.pb.go b/actions/proto/vql.pb.go
diff --git a/artifacts/testdata/server/testcases/query.in.yaml b/artifacts/testdata/server/testcases/query.in.yaml
@@ -0,0 +1,6 @@
+Queries:
+# Exit query with callback
+- |
+  SELECT * FROM query(query={
+    SELECT _value AS Value FROM range(end=100)
+  }, exit="x=>x.Value >= 3")
diff --git a/artifacts/testdata/server/testcases/query.out.yaml b/artifacts/testdata/server/testcases/query.out.yaml
@@ -0,0 +1,17 @@
+# Exit query with callback
+Query: SELECT * FROM query(query={
+  SELECT _value AS Value FROM range(end=100)
+}, exit="x=>x.Value >= 3")
+
+Output: [
+ {
+  "Value": 0
+ },
+ {
+  "Value": 1
+ },
+ {
+  "Value": 2
+ }
+]
+
diff --git a/bin/debian.go b/bin/debian.go
@@ -121,7 +121,12 @@ func doServerDeb() error {
                        directory_name=Output,
                        release=Release)`
 
-	return runQueryWithEnv(query, builder, "json")
+	err = runQueryWithEnv(query, builder, "json")
+	if err != nil {
+		return err
+	}
+
+	return logger.Error
 }
 
 func doClientDeb() error {
@@ -181,7 +186,12 @@ func doClientDeb() error {
                        directory_name=Output,
                        release=Release)`
 
-	return runQueryWithEnv(query, builder, "json")
+	err = runQueryWithEnv(query, builder, "json")
+	if err != nil {
+		return err
+	}
+
+	return logger.Error
 }
 
 func init() {
diff --git a/bin/fuse_unix.go b/bin/fuse_unix.go
@@ -153,7 +153,7 @@ func doFuseZip() error {
 
 	<-ctx.Done()
 
-	return nil
+	return logger.Error
 }
 
 func init() {
diff --git a/bin/offline_decrypt.go b/bin/offline_decrypt.go
@@ -125,5 +125,5 @@ FROM stat(filename=copy(
 		fmt.Printf("Password is: %v\n", password)
 	}
 
-	return nil
+	return logger.Error
 }
diff --git a/bin/rpm.go b/bin/rpm.go
@@ -85,7 +85,7 @@ func doClientRPM() error {
 	if *rpm_command_release == "" {
 		*rpm_command_release = "A"
 	}
-	
+
 	logger := &LogWriter{config_obj: sm.Config}
 	builder := services.ScopeBuilder{
 		Config:     sm.Config,
@@ -106,7 +106,12 @@ func doClientRPM() error {
                        release=Release)
 `
 
-	return runQueryWithEnv(query, builder, "json")
+	err = runQueryWithEnv(query, builder, "json")
+	if err != nil {
+		return err
+	}
+
+	return logger.Error
 }
 
 // Systemd based start up scripts (CentOS 7+)
@@ -146,12 +151,12 @@ func doServerRPM() error {
 	if *server_rpm_command_output == "" {
 		*server_rpm_command_output = "."
 	}
-	
+
 	// By default it should be set to A
 	if *rpm_command_release == "" {
 		*rpm_command_release = "A"
 	}
-	
+
 	logger := &LogWriter{config_obj: sm.Config}
 	builder := services.ScopeBuilder{
 		Config:     sm.Config,
@@ -172,7 +177,12 @@ func doServerRPM() error {
                        release=Release)
 `
 
-	return runQueryWithEnv(query, builder, "json")
+	err = runQueryWithEnv(query, builder, "json")
+	if err != nil {
+		return err
+	}
+
+	return logger.Error
 }
 
 func init() {
diff --git a/config/proto/config.pb.go b/config/proto/config.pb.go
diff --git a/config/proto/config.proto b/config/proto/config.proto
@@ -1175,6 +1175,7 @@ message Defaults {
 
     // How often to refresh the hunt index (default 600 sec)
     int64 hunt_dispatcher_refresh_sec = 43;
+    uint64 hunt_dispatcher_refresh_rate = 60;
 
     // By default new cells only list 50 rows if there is no custom
     // cell template. This is to make refreshing the cell
diff --git a/docs/references/vql.yaml b/docs/references/vql.yaml
@@ -7879,6 +7879,7 @@
     The tracker has two queries: a sync_query and an update_query. The update
     query resets the internal database.
   type: Function
+  version: 2
   args:
   - name: sync_query
     type: StoredQuery
@@ -7891,13 +7892,19 @@
     type: StoredQuery
     description: An Event query that produces live updates of the tracker state.
   - name: max_size
-    type: int64
+    type: uint64
     description: Maximum size of process tracker LRU.
+  - name: max_expiry
+    type: uint64
+    description: Expire process records older than this much.
   - name: enrichments
     type: string
     description: One or more VQL lambda functions that can enrich the data for the
       process.
     repeated: true
+  - name: cache
+    type: string
+    description: The path to the cache file - if not set we use a memory based cache.
   platforms:
   - darwin_amd64_cgo
   - darwin_arm64_cgo
@@ -11838,7 +11845,7 @@
 
     ## Tracee policies.
 
-    As of release 0.76, when calling this plugins, callers can supply
+    As of release 0.76, when calling this plugin, callers can supply
     a tracee policy instead of a list of events. The policy is a YAML
     file in a format described
     [here](https://aquasecurity.github.io/tracee/v0.14/docs/policies/)
@@ -12852,3 +12859,4 @@
   - linux_amd64_cgo
   - windows_386_cgo
   - windows_amd64_cgo
+
diff --git a/gui/velociraptor/src/components/artifacts/artifacts-upload.css b/gui/velociraptor/src/components/artifacts/artifacts-upload.css
@@ -3,6 +3,9 @@
     overflow-y: auto;
 }
 
+.hidden-file-upload {
+    display: none;
+}
 
 #upload {
     display: none;
diff --git a/gui/velociraptor/src/components/artifacts/artifacts-upload.jsx b/gui/velociraptor/src/components/artifacts/artifacts-upload.jsx
@@ -113,6 +113,9 @@ export default class ArtifactsUpload extends React.Component {
                      this.setState({loading:false,
                                     vfs_path: response.data.vfs_path,
                                     uploaded: uploaded});
+                 }).catch(err=>{
+                     console.log(err);
+                     this.setState({loading: false});
                  });
     };
 
@@ -127,10 +130,14 @@ export default class ArtifactsUpload extends React.Component {
         };
         api.post("v1/LoadArtifactPack", request,
                  this.source.token).then(response => {
+                     console.log(response);
+
                      if (response.data.cancel) {
                          return ;
                      }
                      this.props.onClose();
+                 }).catch(response=>{
+                     this.setState({loading: false});
                  });
     };
 
@@ -152,16 +159,16 @@ export default class ArtifactsUpload extends React.Component {
                     <Form.Group as={Row}>
                       <Col sm="12">
                         <InputGroup className="full-width custom-file-button">
-                          <Button variant="default"
-                            className={classNames({
-                                "disabled": !this.state.pack_file
-                            })}
-                            onClick={()=>this.uploadFile()}>
-                            { this.state.loading ?
-                              <FontAwesomeIcon icon="spinner" spin/> :
-                              T("Click to Upload") }
-                          </Button>
+                          { this.state.pack_file &&
+                            <Button variant="default"
+                                    onClick={()=>this.uploadFile()}>
+                              { this.state.loading ?
+                                <FontAwesomeIcon icon="spinner" spin/> :
+                                T("Click to Upload") }
+                            </Button> }
                           <Form.Control type="file" id={this.state.id}
+                                        className="hidden-file-upload"
+                                        placeholder={T("Select a file")}
                                         onChange={e => {
                                             if (!_.isEmpty(e.currentTarget.files)) {
                                                 this.setState({
@@ -173,7 +180,7 @@ export default class ArtifactsUpload extends React.Component {
                           <ToolTip tooltip={T("Select artifact pack (Zip file with YAML definitions)")}>
                             <Button variant="default-outline"
                                     className="flush-right">
-                              <Form.Label data-browse="Select file" htmlFor="upload">
+                              <Form.Label data-browse="Select file" htmlFor={this.state.id}>
                                 {this.state.pack_file ? this.state.pack_file.name :
                                  T("Select artifact pack (Zip file with YAML definitions)")}
                               </Form.Label>
diff --git a/services/hunt_dispatcher/flows.go b/services/hunt_dispatcher/flows.go
@@ -23,7 +23,8 @@ func syncFlowTables(
 	config_obj *config_proto.Config,
 	launcher services.Launcher,
 	hunt_id string,
-	refresh_stats *HuntRefreshStats) (*api_proto.HuntStats, error) {
+	refresh_stats *HuntRefreshStats,
+	throttler *utils.Throttler) (*api_proto.HuntStats, error) {
 
 	// Update the stats if needed.
 	stats := &api_proto.HuntStats{}
@@ -48,7 +49,7 @@ func syncFlowTables(
 
 		// Skip refreshing the enriched table if it is newer than 5 min
 		// old - this helps to reduce unnecessary updates.
-		if now.Sub(enriched_reader.MTime()) < HuntDispatcherRefreshSec(config_obj) {
+		if now.Sub(enriched_reader.MTime()) < HuntDispatcherRefresh(config_obj) {
 			return nil, utils.CancelledError
 		}
 	}
@@ -67,6 +68,10 @@ func syncFlowTables(
 	}
 
 	for json_str := range json_chan {
+		if throttler != nil {
+			throttler.Wait()
+		}
+
 		participation_row := &hunt_manager.ParticipationRecord{}
 		err := json.Unmarshal(json_str, participation_row)
 		if err != nil {
@@ -140,7 +145,7 @@ func (self *HuntDispatcher) GetFlows(
 	// the original table.
 	if options.SortColumn != "" || options.FilterColumn != "" {
 		_, err := syncFlowTables(ctx, config_obj, launcher, hunt_id,
-			&HuntRefreshStats{})
+			&HuntRefreshStats{}, nil)
 		if err != nil && !errors.Is(err, utils.CancelledError) {
 			close(output_chan)
 			return output_chan, 0, err
diff --git a/services/hunt_dispatcher/hunt_dispatcher.go b/services/hunt_dispatcher/hunt_dispatcher.go
@@ -399,7 +399,7 @@ func (self *HuntDispatcher) StartRefresh(
 			return
 		}
 
-		refresh := HuntDispatcherRefreshSec(config_obj)
+		refresh := HuntDispatcherRefresh(config_obj)
 
 		logger := logging.GetLogger(config_obj, &logging.FrontendComponent)
 		logger.Info("<green>Starting</> Hunt Dispatcher Service for %v.",
@@ -491,13 +491,14 @@ func GetNewHuntId() string {
 	return constants.HUNT_PREFIX + result
 }
 
-func HuntDispatcherRefreshSec(config_obj *config_proto.Config) time.Duration {
+func HuntDispatcherRefresh(config_obj *config_proto.Config) time.Duration {
 	if config_obj.Defaults != nil &&
 		config_obj.Defaults.HuntDispatcherRefreshSec > 0 {
 		return time.Duration(config_obj.Defaults.HuntDispatcherRefreshSec) * time.Second
 	}
 
-	return time.Minute
+	// By default refresh every 10 minutes
+	return 10 * time.Minute
 }
 
 func init() {
diff --git a/services/hunt_dispatcher/storage.go b/services/hunt_dispatcher/storage.go
@@ -113,15 +113,26 @@ type HuntStorageManagerImpl struct {
 	last_flush_time time.Time
 
 	tracker *HuntDispatcherTracker
+
+	refresh_throttler *utils.Throttler
 }
 
 func NewHuntStorageManagerImpl(
 	config_obj *config_proto.Config) HuntStorageManager {
+
+	// To limit CPU and IO we throttle the refresh operation
+	refresh_rate := uint64(100)
+	if config_obj.Defaults != nil &&
+		config_obj.Defaults.HuntDispatcherRefreshRate > 0 {
+		refresh_rate = config_obj.Defaults.HuntDispatcherRefreshRate
+	}
+
 	result := &HuntStorageManagerImpl{
-		config_obj:  config_obj,
-		hunts:       make(map[string]*HuntRecord),
-		I_am_master: services.IsMaster(config_obj),
-		tracker:     &HuntDispatcherTracker{},
+		config_obj:        config_obj,
+		hunts:             make(map[string]*HuntRecord),
+		I_am_master:       services.IsMaster(config_obj),
+		tracker:           &HuntDispatcherTracker{},
+		refresh_throttler: utils.NewThrottler(refresh_rate),
 	}
 
 	if result.I_am_master {
@@ -163,6 +174,7 @@ func (self *HuntStorageManagerImpl) Close(ctx context.Context) {
 		logger := logging.GetLogger(self.config_obj, &logging.FrontendComponent)
 		logger.Error("HuntStorageManager FlushIndex %v", err)
 	}
+	self.refresh_throttler.Close()
 }
 
 func (self *HuntStorageManagerImpl) ModifyHuntObject(
@@ -464,7 +476,8 @@ func (self *HuntStorageManagerImpl) loadHuntObjFromDisk(
 
 	// Scan the client list to update the scheduled and errored count.
 	stats, err := syncFlowTables(
-		ctx, config_obj, launcher, hunt_obj.HuntId, refresh_stats)
+		ctx, config_obj, launcher, hunt_obj.HuntId,
+		refresh_stats, self.refresh_throttler)
 	if err == nil && isStatsUpdated(stats, hunt_obj.Stats) {
 		hunt_obj.Stats = stats
 		if hunt_obj.State == api_proto.Hunt_STOPPED {
@@ -493,6 +506,8 @@ func (self *HuntStorageManagerImpl) loadHuntsFromDatastore(
 		Time: utils.GetTime().Now(),
 	}
 
+	self.tracker.AddRefreshStats(stats)
+
 	// Ensure all the records are ready to read.
 	err := datastore.FlushDatastore(config_obj)
 	if err != nil {
@@ -580,8 +595,9 @@ func (self *HuntStorageManagerImpl) loadHuntsFromDatastore(
 		self.last_update = utils.GetTime().Now()
 	}
 
+	stats.Lock()
 	stats.Duration = utils.GetTime().Now().Sub(stats.Time)
-	self.tracker.AddRefreshStats(stats)
+	stats.Unlock()
 
 	return nil
 }
diff --git a/services/hunt_dispatcher/tracker.go b/services/hunt_dispatcher/tracker.go
diff --git a/vql/linux/ebpf/ebpf.go b/vql/linux/ebpf/ebpf.go
diff --git a/vql/tools/query.go b/vql/tools/query.go
diff --git a/vql/tools/template.go b/vql/tools/template.go

Original file line number	Diff line number	Diff line change
`@@ -153,7 +153,7 @@ func doFuseZip() error {`
`153`	`153`
`154`	`154`	`<-ctx.Done()`
`155`	`155`
`156`		`- return nil`
	`156`	`+ return logger.Error`
`157`	`157`	`}`
`158`	`158`
`159`	`159`	`func init() {`
Original file line number	Diff line number	Diff line change
`@@ -125,5 +125,5 @@ FROM stat(filename=copy(`
`125`	`125`	`fmt.Printf("Password is: %v\n", password)`
`126`	`126`	`}`
`127`	`127`
`128`		`- return nil`
	`128`	`+ return logger.Error`
`129`	`129`	`}`