Merge pull request #810 from forta-network/caner/forta-1277-report-do…

…cker-events-from-nodes

Report Docker events as metrics

Makefile

@@ -1,11 +1,9 @@
 containers:
 	docker build -t forta-network/forta-node -f Dockerfile.node .
-	docker pull nats:2.3.2
 
 containers-dev:
 	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o forta-node cmd/node/main.go
 	DOCKER_BUILDKIT=1 docker build --no-cache --network=host -t forta-network/forta-node -f Dockerfile.buildkit.dev.node .
-	docker pull nats:2.3.2
 
 main:
 	docker build -t build-forta -f Dockerfile.cli .

clients/docker/client.go

@@ -12,16 +12,16 @@ import (
 	"strings"
 	"time"
 
-	"github.com/goccy/go-json"
-
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/events"
 	"github.com/docker/docker/api/types/filters"
 	"github.com/docker/docker/client"
 	"github.com/docker/go-connections/nat"
 	"github.com/forta-network/forta-core-go/utils/workers"
 	"github.com/forta-network/forta-node/clients/cooldown"
 	"github.com/forta-network/forta-node/config"
+	"github.com/goccy/go-json"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -843,6 +843,13 @@ func makeLabelFilter(labels []dockerLabel) filters.Args {
 	return filter
 }
 
+// Events returns channels that send the Docker events and listening/decoding errors.
+func (d *dockerClient) Events(ctx context.Context, since time.Time) (<-chan events.Message, <-chan error) {
+	return d.cli.Events(ctx, types.EventsOptions{
+		Since: since.Format(time.RFC3339),
+	})
+}
+
 func (d *dockerClient) GetContainerFromRemoteAddr(ctx context.Context, hostPort string) (*types.Container, error) {
 	containers, err := d.GetContainers(ctx)
 	if err != nil {

clients/interfaces.go

@@ -4,13 +4,12 @@ import (
 	"context"
 	"time"
 
-	"github.com/forta-network/forta-core-go/domain"
-
 	"github.com/docker/docker/api/types"
-	"github.com/golang/protobuf/proto"
-
+	"github.com/docker/docker/api/types/events"
+	"github.com/forta-network/forta-core-go/domain"
 	"github.com/forta-network/forta-node/clients/docker"
 	"github.com/forta-network/forta-node/config"
+	"github.com/golang/protobuf/proto"
 )
 
 // DockerClient is a client interface for interacting with docker
@@ -47,6 +46,7 @@ type DockerClient interface {
 	GetContainerLogs(ctx context.Context, containerID, tail string, truncate int) (string, error)
 	GetContainerFromRemoteAddr(ctx context.Context, hostPort string) (*types.Container, error)
 	SetImagePullCooldown(threshold int, cooldownDuration time.Duration)
+	Events(ctx context.Context, since time.Time) (<-chan events.Message, <-chan error)
 }
 
 // MessageClient receives and publishes messages.

services/components/containers/definitions.go

@@ -15,7 +15,7 @@ const (
 	LabelValueFortaIsBot = "true"
 	// LabelValueStrategyVersion is for versioning the critical changes in container management strategy.
 	// It's effective in deciding if a bot container should be re-created or not.
-	LabelValueStrategyVersion = "2023-06-16T15:00:00Z"
+	LabelValueStrategyVersion = "2023-09-20T12:00:00Z"
 )
 
 // Limits define container limits.

services/components/containers/events.go

@@ -0,0 +1,117 @@
+package containers
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"github.com/docker/docker/api/types/events"
+	"github.com/forta-network/forta-core-go/protocol"
+	"github.com/forta-network/forta-node/clients"
+	"github.com/forta-network/forta-node/clients/docker"
+	"github.com/forta-network/forta-node/services/components/metrics"
+	"github.com/sirupsen/logrus"
+)
+
+// ListenToDockerEvents creates new.
+func ListenToDockerEvents(
+	ctx context.Context, dockerClient clients.DockerClient, msgClient clients.MessageClient,
+	startFrom time.Time,
+) {
+	handler := &eventHandler{
+		dockerClient: dockerClient,
+		msgClient:    msgClient,
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			logrus.Info("stopping docker events listener")
+			return
+		default:
+		}
+
+		events, errs := dockerClient.Events(ctx, startFrom)
+
+		var restartListening bool
+		for {
+			select {
+			case event := <-events:
+				handler.HandleEvent(ctx, &event)
+
+			case err := <-errs:
+				logrus.WithError(err).Error("error while listening to docker events")
+				// set the start time and restart listening
+				startFrom = time.Now().Add(-1 * time.Second)
+				restartListening = true
+
+			case <-ctx.Done():
+				logrus.Info("stopping docker events listener")
+				return
+			}
+			if restartListening {
+				break
+			}
+		}
+	}
+}
+
+type eventHandler struct {
+	dockerClient clients.DockerClient
+	msgClient    clients.MessageClient
+}
+
+func (es *eventHandler) HandleEvent(ctx context.Context, event *events.Message) {
+	var metric *protocol.AgentMetric
+	ts := time.Unix(0, event.TimeNano)
+	switch event.Type {
+	case "image":
+		if event.Action != "pull" {
+			return
+		}
+		imageRef := getEventAttribute(event, "name")
+		metric = metrics.CreateEventMetric(ts, "system", metricNameFrom(event), imageRef)
+
+	case "container", "network":
+		if !isOneOf(event.Action, "create", "destroy", "connect", "disconnect") {
+			return
+		}
+		botID, ok := getBotID(event)
+		if !ok {
+			botID = "system"
+		}
+		containerName := getEventAttribute(event, "name")
+		metric = metrics.CreateEventMetric(ts, botID, metricNameFrom(event), containerName)
+
+	default:
+		return
+	}
+
+	metrics.SendAgentMetrics(es.msgClient, []*protocol.AgentMetric{metric})
+	return
+}
+
+func isOneOf(input string, values ...string) bool {
+	for _, value := range values {
+		if input == value {
+			return true
+		}
+	}
+	return false
+}
+
+func metricNameFrom(event *events.Message) string {
+	return strings.Join([]string{"docker", event.Type, event.Action}, ".")
+}
+
+func getBotID(event *events.Message) (string, bool) {
+	val := getEventAttribute(event, docker.LabelFortaBotID)
+	return val, len(val) > 0
+}
+
+func getEventAttribute(event *events.Message, attr string) string {
+	if event.Actor.Attributes == nil {
+		return ""
+	}
+	return event.Actor.Attributes[attr]
+}