Notifications are throttled now. Refactor VQLCollectorArgs compiler. (V…

…elocidex#958)

This change controls the rate of notifications in the notifications
service. Usually when a new hunt is created, the notifier notifies all
the clients immediately so they can re-connect and receive the next
hunt. On very large deployments this causes a bottleneck and a slow
down of the server.

To address this issue we
1. Limit the rate of notifications to ensure not all clients are
   reconnecting at the same time.
2. Limit the total connection rate to ensure the server is not
   overwhelmed. This simply returns 500 when the connection rate is
   exceeded.

Additionally this PR changes the way tools are passed to the
VQLCollectorArgs when the artifact is compiled. Previously tool
information was passed in the global artifact's
env (e.g. Tool_XXX_HASH), and we relies on the previous behaviour that
all dependent artifacts shared the root scope.

However in recent releases dependent artifacts are running in an
isolated scope and so can not see the tool information. This change
sets the tool information as a parameter in each dependent artifact's
definition to ensure that tool is properly resoved on the client.

artifacts/definitions/Windows/NTFS/Recover.yaml

@@ -13,7 +13,7 @@ description: |
 
 parameters:
  - name: MFTId
-   default: 81978
+   default: "81978"
  - name: Drive
    default: '\\.\C:'
 
@@ -23,11 +23,17 @@ precondition:
 sources:
   - name: Upload
     queries:
-       - SELECT * FROM foreach(
+       - SELECT *, upload(accessor="mft", file=Drive + Inode,
+                          name=FullPath + "/" + Inode) AS IndexUpload
+         FROM foreach(
             row=parse_ntfs(device=Drive, inode=MFTId).Attributes,
             query={
-              SELECT Type, TypeId, Id, Inode, Size, Name, FullPath,
-                     upload(accessor="mft", file=Drive + Inode,
-                            name=FullPath + "/" + Inode) AS IndexUpload
+              SELECT _value.Type AS Type,
+                     _value.TypeId AS TypeId,
+                     _value.Id AS Id,
+                     _value.Inode AS Inode,
+                     _value.Size AS Size,
+                     _value.Name AS Name,
+                     _value.FullPath AS FullPath
               FROM scope()
             })

config/config.go

@@ -133,15 +133,17 @@ func GetDefaultConfig() *config_proto.Config {
 
 			// A public interface for clients to
 			// connect to.
-			BindAddress:   "0.0.0.0",
-			BindPort:      8000,
-			MaxUploadSize: constants.MAX_MEMORY * 2,
+			BindAddress: "0.0.0.0",
+			BindPort:    8000,
 			DefaultClientMonitoringArtifacts: []string{
 				// Essential for client resource telemetry.
 				"Generic.Client.Stats",
 			},
-			DynDns:          &config_proto.DynDNSConfig{},
-			ExpectedClients: 10000,
+			DynDns: &config_proto.DynDNSConfig{},
+			Resources: &config_proto.FrontendResourceControl{
+				ExpectedClients: 10000,
+				MaxUploadSize:   constants.MAX_MEMORY * 2,
+			},
 			GRPCPoolMaxSize: 100,
 			GRPCPoolMaxWait: 60,
 		},

config/migration.go

@@ -133,6 +133,19 @@ func migrate_0_5_6(config_obj *config_proto.Config) {
 			config_obj.Logging.Debug = default_rotator
 		}
 	}
+
+	if config_obj.Frontend != nil {
+		if config_obj.Frontend.Resources == nil {
+			config_obj.Frontend.Resources = &config_proto.FrontendResourceControl{
+				Concurrency:         config_obj.Frontend.Concurrency,
+				MaxUploadSize:       config_obj.Frontend.MaxUploadSize,
+				ExpectedClients:     config_obj.Frontend.ExpectedClients,
+				PerClientUploadRate: config_obj.Frontend.PerClientUploadRate,
+				GlobalUploadRate:    config_obj.Frontend.GlobalUploadRate,
+				ClientEventMaxWait:  config_obj.Frontend.ClientEventMaxWait,
+			}
+		}
+	}
 }
 
 func migrate(config_obj *config_proto.Config) {