Monitor the Nomad events

and send all Nomad events to Influxdb.

internal/nomad/api_querier.go

@@ -181,6 +181,9 @@ func (nc *nomadAPIClient) EventStream(ctx context.Context) (<-chan *nomadApi.Eve
 				// As Poseidon uses no such token, the request will return a permission denied error.
 				"*",
 			},
+			nomadApi.TopicJob:        {"*"},
+			nomadApi.TopicNode:       {"*"},
+			nomadApi.TopicDeployment: {"*"},
 		},
 		0,
 		nc.queryOptions())

internal/nomad/nomad.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	nomadApi "github.com/hashicorp/nomad/api"
 	"github.com/hashicorp/nomad/nomad/structs"
+	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
 	"github.com/influxdata/influxdb-client-go/v2/api/write"
 	"github.com/openHPI/poseidon/internal/config"
 	"github.com/openHPI/poseidon/pkg/dto"
@@ -207,6 +208,7 @@ func (a *APIClient) WatchEventStream(ctx context.Context, callbacks *AllocationP
 	}
 
 	handler := func(event *nomadApi.Event) (bool, error) {
+		dumpNomadEventToInflux(event)
 		switch event.Topic {
 		case nomadApi.TopicEvaluation:
 			return false, handleEvaluationEvent(a.evaluations, event)
@@ -223,6 +225,15 @@ func (a *APIClient) WatchEventStream(ctx context.Context, callbacks *AllocationP
 	return err
 }
 
+func dumpNomadEventToInflux(event *nomadApi.Event) {
+	p := influxdb2.NewPointWithMeasurement(monitoring.MeasurementNomadEvents)
+	p.AddTag("topic", event.Topic.String())
+	p.AddTag("type", event.Type)
+	p.AddTag("key", event.Key)
+	p.AddField("payload", event.Payload)
+	monitoring.WriteInfluxPoint(p)
+}
+
 func (a *APIClient) initializeAllocations() {
 	allocationStubs, err := a.listAllocations()
 	if err != nil {