Push handler metrics to main service when exiting (#565)

This PR prevents the last prometheus metrics to be lost at the end of an egress by pushing them in a new ipc message to the main service process. This is particularly important since the last set of metrics contains file and manifest upload information.

The metrics pushed by the handler on exit are stored in an array in the Service object. They are appended to the other metrics in the prometheus gatherer.

pkg/handler/handler.go

@@ -21,6 +21,9 @@ import (
 	"time"
 
 	"github.com/frostbyte73/core"
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/prometheus/common/expfmt"
 	"google.golang.org/grpc"
 
 	"github.com/livekit/egress/pkg/config"
@@ -120,6 +123,17 @@ func (h *Handler) Run() error {
 		case res := <-result:
 			// recording finished
 			_, _ = h.ioClient.UpdateEgress(ctx, res)
+
+			m, err := h.GenerateMetrics(ctx)
+			if err == nil {
+				h.ipcServiceClient.HandlerShuttingDown(ctx, &ipc.HandlerShuttingDownRequest{
+					EgressId: h.conf.Info.EgressId,
+					Metrics:  m,
+				})
+			} else {
+				logger.Errorw("failed generating handler metrics", err)
+			}
+
 			h.rpcServer.Shutdown()
 			h.ipcHandlerServer.Stop()
 			return nil
@@ -130,3 +144,35 @@ func (h *Handler) Run() error {
 func (h *Handler) Kill() {
 	h.kill.Break()
 }
+
+func (h *Handler) GenerateMetrics(ctx context.Context) (string, error) {
+	metrics, err := prometheus.DefaultGatherer.Gather()
+	if err != nil {
+		return "", err
+	}
+
+	metricsAsString, err := renderMetrics(metrics)
+	if err != nil {
+		return "", err
+	}
+
+	return metricsAsString, nil
+}
+
+func renderMetrics(metrics []*dto.MetricFamily) (string, error) {
+	// Create a StringWriter to render the metrics into text format
+	writer := &strings.Builder{}
+	totalCnt := 0
+	for _, metric := range metrics {
+		// Write each metric family to text
+		cnt, err := expfmt.MetricFamilyToText(writer, metric)
+		if err != nil {
+			logger.Errorw("error writing metric family", err)
+			return "", err
+		}
+		totalCnt += cnt
+	}
+
+	// Get the rendered metrics as a string from the StringWriter
+	return writer.String(), nil
+}

pkg/handler/handler_ipc.go

@@ -16,18 +16,13 @@ package handler
 
 import (
 	"context"
-	"strings"
 	"time"
 
-	"github.com/prometheus/client_golang/prometheus"
-	dto "github.com/prometheus/client_model/go"
-	"github.com/prometheus/common/expfmt"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
 	"github.com/livekit/egress/pkg/errors"
 	"github.com/livekit/egress/pkg/ipc"
-	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/pprof"
 	"github.com/livekit/protocol/tracer"
 )
@@ -79,12 +74,7 @@ func (h *Handler) GetMetrics(ctx context.Context, req *ipc.MetricsRequest) (*ipc
 	ctx, span := tracer.Start(ctx, "Handler.GetMetrics")
 	defer span.End()
 
-	metrics, err := prometheus.DefaultGatherer.Gather()
-	if err != nil {
-		return nil, err
-	}
-
-	metricsAsString, err := renderMetrics(metrics)
+	metricsAsString, err := h.GenerateMetrics(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -93,21 +83,3 @@ func (h *Handler) GetMetrics(ctx context.Context, req *ipc.MetricsRequest) (*ipc
 		Metrics: metricsAsString,
 	}, nil
 }
-
-func renderMetrics(metrics []*dto.MetricFamily) (string, error) {
-	// Create a StringWriter to render the metrics into text format
-	writer := &strings.Builder{}
-	totalCnt := 0
-	for _, metric := range metrics {
-		// Write each metric family to text
-		cnt, err := expfmt.MetricFamilyToText(writer, metric)
-		if err != nil {
-			logger.Errorw("error writing metric family", err)
-			return "", err
-		}
-		totalCnt += cnt
-	}
-
-	// Get the rendered metrics as a string from the StringWriter
-	return writer.String(), nil
-}