More multi-frontend optimizations (Velocidex#1393)

* More multi-frontend optimizations

* Client info manager now also keeps track of LastHuntTimestamp and
  LastEventTableVersion for each client. This means we do not need to
  send an UpdateForeman message to the client since the server keeps
  track

* Client info manager now keeps track of the state of client
  queues. If no new messages are queued for the client, the frontend
  is able to skip making an IO operation on the data store.

* Replication service now writes the events to storage by itself. This
  ensures that minions are able to take on the writing work for event
  queries.

* Replication service now receives an update on all Watchers in the
  master, and avoids sending any events to the master than no watches
  are interested in. This reduces the number of gRPC calls between
  minion and master and therefore improves performance.

actions/proto/vql.proto

@@ -159,10 +159,14 @@ message ClientInfo {
     string architecture = 7;
     string ip_address = 10;
     uint64 ping = 11;
+    string ping_time = 19;
     string client_version = 12;
     string client_name = 13;
 
     repeated string labels = 15;
 
     string last_interrogate_flow_id = 16;
+
+    uint64 last_hunt_timestamp = 17;
+    uint64 last_event_table_version = 18;
 }

api/api.go

@@ -218,13 +218,7 @@ func (self *ApiServer) CollectArtifact(
 	}
 
 	flow_id, err := launcher.ScheduleArtifactCollection(
-		ctx, self.config, acl_manager, repository, in,
-		func() {
-			notifier := services.GetNotifier()
-			if notifier != nil {
-				notifier.NotifyListener(self.config, in.ClientId)
-			}
-		})
+		ctx, self.config, acl_manager, repository, in, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -292,13 +286,10 @@ func (self *ApiServer) NotifyClients(
 		return nil, errors.New("Notifier not ready")
 	}
 
-	if in.NotifyAll {
-		self.server_obj.Info("sending notification to everyone")
-		err = notifier.NotifyAllListeners(self.config)
-
-	} else if in.ClientId != "" {
+	if in.ClientId != "" {
 		self.server_obj.Info("sending notification to %s", in.ClientId)
-		err = services.GetNotifier().NotifyListener(self.config, in.ClientId)
+		err = services.GetNotifier().NotifyListener(
+			self.config, in.ClientId, "API.NotifyClients")
 	} else {
 		return nil, status.Error(codes.InvalidArgument,
 			"client id should be specified")
@@ -841,10 +832,6 @@ func (self *ApiServer) SetClientMonitoringState(
 		return nil, err
 	}
 
-	_, err = self.NotifyClients(ctx, &api_proto.NotificationRequest{
-		NotifyAll: true,
-	})
-
 	return &empty.Empty{}, err
 }
 

api/builder.go

@@ -14,7 +14,6 @@ import (
 	config_proto "www.velocidex.com/golang/velociraptor/config/proto"
 	"www.velocidex.com/golang/velociraptor/logging"
 	"www.velocidex.com/golang/velociraptor/server"
-	"www.velocidex.com/golang/velociraptor/services"
 
 	_ "www.velocidex.com/golang/velociraptor/result_sets/timed"
 )
@@ -426,13 +425,6 @@ func StartFrontendPlainHttp(
 		defer cancel()
 
 		server.SetKeepAlivesEnabled(false)
-		notifier := services.GetNotifier()
-		if notifier != nil {
-			err := notifier.NotifyAllListeners(config_obj)
-			if err != nil {
-				server_obj.Error("Frontend server error %v", err)
-			}
-		}
 		err := server.Shutdown(time_ctx)
 		if err != nil {
 			server_obj.Error("Frontend server error %v", err)
@@ -523,10 +515,6 @@ func StartFrontendWithAutocert(
 		defer cancel()
 
 		server.SetKeepAlivesEnabled(false)
-		notifier := services.GetNotifier()
-		if notifier != nil {
-			_ = notifier.NotifyAllListeners(config_obj)
-		}
 		err := server.Shutdown(timeout_ctx)
 		if err != nil {
 			logger.Error("Frontend shutdown error: %v", err)

api/notebooks.go

@@ -853,13 +853,15 @@ func (self *ApiServer) CancelNotebookCell(
 	}
 
 	notebook_cell.Calculating = false
+	// Make sure we write the cancel message ASAP
 	err = db.SetSubject(self.config, notebook_cell_path_manager.Path(),
 		notebook_cell)
 	if err != nil {
 		return nil, err
 	}
 
-	return &empty.Empty{}, services.GetNotifier().NotifyListener(self.config, in.CellId)
+	return &empty.Empty{}, services.GetNotifier().NotifyListener(
+		self.config, in.CellId, "CancelNotebookCell")
 }
 
 func (self *ApiServer) UploadNotebookAttachment(

api/replication.go

@@ -4,6 +4,9 @@ import (
 	"crypto/x509"
 	"fmt"
 
+	"github.com/Velocidex/ordereddict"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/sirupsen/logrus"
 	context "golang.org/x/net/context"
 	"google.golang.org/grpc/codes"
@@ -19,6 +22,17 @@ import (
 	"www.velocidex.com/golang/velociraptor/services"
 )
 
+var (
+	replicationReceiveHistorgram = promauto.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Name:    "replication_master_send_latency",
+			Help:    "Latency for the master to send replication messages to the minion.",
+			Buckets: prometheus.LinearBuckets(0.1, 1, 10),
+		},
+		[]string{"status"},
+	)
+)
+
 func streamEvents(
 	ctx context.Context,
 	config_obj *config_proto.Config,
@@ -37,6 +51,16 @@ func streamEvents(
 		return err
 	}
 
+	// Special case this so the caller can immediately initialize the
+	// watchers.
+	if in.Queue == "Server.Internal.MasterRegistrations" {
+		result := ordereddict.NewDict().Set("Events", journal.GetWatchers())
+		serialized, _ := result.MarshalJSON()
+		stream.Send(&api_proto.EventResponse{
+			Jsonl: serialized,
+		})
+	}
+
 	// The API service is running on the master only! This means
 	// the journal service is local.
 	output_chan, cancel := journal.Watch(ctx, in.Queue)
@@ -50,7 +74,15 @@ func streamEvents(
 		response := &api_proto.EventResponse{
 			Jsonl: serialized,
 		}
+
+		timer := prometheus.NewTimer(
+			prometheus.ObserverFunc(func(v float64) {
+				replicationReceiveHistorgram.WithLabelValues("").Observe(v)
+			}))
+
 		err = stream.Send(response)
+		timer.ObserveDuration()
+
 		if err != nil {
 			continue
 		}

artifacts/definitions/Server/Internal/ClientTasks.yaml

@@ -0,0 +1,7 @@
+name: Server.Internal.ClientTasks
+description: |
+  This event will be fired when a client has new tasks scheduled.
+
+type: INTERNAL
+column_types:
+  - name: ClientId