Monitor file download.

internal/runner/nomad_runner.go

@@ -9,6 +9,7 @@ import (
 	"errors"
 	"fmt"
 	nomadApi "github.com/hashicorp/nomad/api"
+	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
 	"github.com/openHPI/poseidon/internal/nomad"
 	"github.com/openHPI/poseidon/pkg/dto"
 	"github.com/openHPI/poseidon/pkg/monitoring"
@@ -190,6 +191,15 @@ func (r *NomadJob) GetFileContent(
 	r.ResetTimeout()
 
 	contentLengthWriter := &nullio.ContentLengthWriter{Target: content}
+	p := influxdb2.NewPointWithMeasurement(monitoring.MeasurementFileDownload)
+	p.AddTag(monitoring.InfluxKeyRunnerID, r.ID())
+	environmentID, err := nomad.EnvironmentIDFromRunnerID(r.ID())
+	if err != nil {
+		log.WithField("runnerID", r.ID()).WithError(err).Warn("can not parse environment id")
+	}
+	p.AddTag(monitoring.InfluxKeyEnvironmentID, environmentID.ToString())
+	defer contentLengthWriter.SendMonitoringData(p)
+
 	retrieveCommand := (&dto.ExecutionRequest{
 		Command: fmt.Sprintf("%s %q && cat %q", lsCommand, path, path),
 	}).FullCommand()

pkg/monitoring/influxdb2_middleware.go

@@ -28,11 +28,14 @@ const (
 	MeasurementExecutionsNomad = measurementPrefix + "nomad_executions"
 	MeasurementEnvironments    = measurementPrefix + "environments"
 	MeasurementUsedRunner      = measurementPrefix + "used_runners"
+	MeasurementFileDownload    = measurementPrefix + "file_download"
 
 	// The keys for the monitored tags and fields.
 
 	InfluxKeyRunnerID                      = "runner_id"
 	InfluxKeyEnvironmentID                 = "environment_id"
+	InfluxKeyActualContentLength           = "actual_length"
+	InfluxKeyExpectedContentLength         = "expected_length"
 	InfluxKeyDuration                      = "duration"
 	InfluxKeyStartupDuration               = "startup_" + InfluxKeyDuration
 	influxKeyEnvironmentPrewarmingPoolSize = "prewarming_pool_size"

pkg/nullio/content_length.go

@@ -3,7 +3,10 @@ package nullio
 import (
 	"errors"
 	"fmt"
+	"github.com/influxdata/influxdb-client-go/v2/api/write"
+	"github.com/openHPI/poseidon/pkg/monitoring"
 	"net/http"
+	"strconv"
 )
 
 var ErrRegexMatching = errors.New("could not match content length")
@@ -15,17 +18,28 @@ type ContentLengthWriter struct {
 	Target                http.ResponseWriter
 	contentLengthSet      bool
 	firstLine             []byte
+	expectedContentLength int
+	actualContentLength   int
 }
 
 func (w *ContentLengthWriter) Write(p []byte) (count int, err error) {
 	if w.contentLengthSet {
-		count, err = w.Target.Write(p)
+		return w.handleDataForwarding(p)
+	} else {
+		return w.handleContentLengthParsing(p)
+	}
+}
+
+func (w *ContentLengthWriter) handleDataForwarding(p []byte) (int, error) {
+	count, err := w.Target.Write(p)
 	if err != nil {
 		err = fmt.Errorf("could not write to target: %w", err)
 	}
+	w.actualContentLength += count
 	return count, err
-	}
+}
 
+func (w *ContentLengthWriter) handleContentLengthParsing(p []byte) (count int, err error) {
 	for i, char := range p {
 		if char != '\n' {
 			continue
@@ -38,6 +52,10 @@ func (w *ContentLengthWriter) Write(p []byte) (count int, err error) {
 			return 0, ErrRegexMatching
 		}
 		size := string(matches[headerLineGroupSize])
+		w.expectedContentLength, err = strconv.Atoi(size)
+		if err != nil {
+			log.WithField("size", size).Warn("could not parse content length")
+		}
 		w.Target.Header().Set("Content-Length", size)
 		w.contentLengthSet = true
 
@@ -57,3 +75,10 @@ func (w *ContentLengthWriter) Write(p []byte) (count int, err error) {
 
 	return len(p), nil
 }
+
+// SendMonitoringData will send a monitoring event of the content length read and written.
+func (w *ContentLengthWriter) SendMonitoringData(p *write.Point) {
+	p.AddField(monitoring.InfluxKeyExpectedContentLength, w.expectedContentLength)
+	p.AddField(monitoring.InfluxKeyActualContentLength, w.actualContentLength)
+	monitoring.WriteInfluxPoint(p)
+}