USHIFT-5297: Nits

Author: pacevedom

pkg/controllers/clusterid.go

@@ -98,3 +98,11 @@ func initClusterIDFile(clusterID string) error {
 	klog.Infof("Writing MicroShift Cluster ID '%v' to '%v'", clusterID, ClusterIDFilePath)
 	return os.WriteFile(ClusterIDFilePath, []byte(clusterID), 0400)
 }
+
+func GetClusterId() (string, error) {
+	data, err := os.ReadFile(ClusterIDFilePath)
+	if err != nil {
+		return "", fmt.Errorf("failed to read file: %v", err)
+	}
+	return string(data), nil
+}

pkg/controllers/telemetry.go

@@ -49,58 +49,51 @@ func (t *TelemetryManager) Dependencies() []string {
 }
 
 func (t *TelemetryManager) Run(ctx context.Context, ready chan<- struct{}, stopped chan<- struct{}) error {
+	// The service manager expects a service to get ready eventually before a timeout or else
+	// MicroShift fails to start correctly. Stopping a service is relevant only when stopping
+	// MicroShift, therefore if Telemetry is disabled we need to signal a fake readiness (even
+	// though it is technically correct, the service is ready even if it is not going to do anything).
+	// For stoppin the service we need to close the stopped channel or else MicroShift fails
+	// to stop gracefully.
 	defer close(stopped)
-	defer close(ready)
-
 	if t.config.Telemetry.Status == config.StatusDisabled {
 		klog.Info("Telemetry is disabled")
+		close(ready)
 		return nil
 	}
-
-	clusterId, err := getClusterId()
+	clusterId, err := GetClusterId()
 	if err != nil {
+		close(ready)
 		return fmt.Errorf("unable to get cluster id: %v", err)
 	}
-
-	go func() {
-		client := telemetry.NewTelemetryClient(t.config.Telemetry.Endpoint, clusterId)
-		collectAndSend := func() {
-			pullSecret, err := readPullSecret()
-			if err != nil {
-				klog.Errorf("Unable to get pull secret: %v", err)
-				return
-			}
-			//TODO swap with collected metrics from client.
-			metrics := []telemetry.Metric{}
-			if err := client.Send(ctx, pullSecret, metrics); err != nil {
-				klog.Errorf("Failed to send metrics: %v", err)
-			}
+	close(ready)
+	client := telemetry.NewTelemetryClient(t.config.Telemetry.Endpoint, clusterId)
+	collectAndSend := func() {
+		pullSecret, err := readPullSecret()
+		if err != nil {
+			klog.Errorf("Unable to get pull secret: %v", err)
+			return
 		}
-
-		klog.Infof("First metrics collection")
-		collectAndSend()
-
-		for {
-			select {
-			case <-ctx.Done():
-				klog.Infof("collect and send for the last time")
-				collectAndSend()
-				return
-			case <-time.After(time.Hour):
-				klog.Infof("collect and send again")
-				collectAndSend()
-			}
+		//TODO swap with collected metrics from client.
+		metrics := []telemetry.Metric{}
+		if err := client.Send(ctx, pullSecret, metrics); err != nil {
+			klog.Errorf("Failed to send metrics: %v", err)
 		}
-	}()
-	return nil
-}
+	}
 
-func getClusterId() (string, error) {
-	data, err := os.ReadFile(ClusterIDFilePath)
-	if err != nil {
-		return "", fmt.Errorf("failed to read file: %v", err)
+	klog.Infof("First metrics collection")
+	collectAndSend()
+	for {
+		select {
+		case <-ctx.Done():
+			klog.Infof("Collect and send for the last time")
+			collectAndSend()
+			return nil
+		case <-time.After(time.Hour):
+			klog.Infof("Collect and send again")
+			collectAndSend()
+		}
 	}
-	return string(data), nil
 }
 
 func readPullSecret() (string, error) {

pkg/telemetry/telemetry.go

@@ -89,12 +89,15 @@ func (t *TelemetryClient) Send(ctx context.Context, pullSecret string, metrics [
 		return fmt.Errorf("unable to do the request: %v", err)
 	}
 	defer func() {
+		// Discard the body to close the TCP socket right away instead of waiting for
+		// the timeout in TIME_WAIT.
 		if _, err := io.Copy(io.Discard, resp.Body); err != nil {
 			klog.Error(err, "error discarding body")
 		}
 		resp.Body.Close()
 	}()
 	if resp.StatusCode == http.StatusOK {
+		klog.Infof("Metrics sent successfully")
 		return nil
 	}
 	body, err := io.ReadAll(resp.Body)

pkg/telemetry/telemetry_test.go

@@ -0,0 +1,85 @@
+package telemetry
+
+import (
+	"context"
+	"encoding/base64"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"testing"
+	"time"
+
+	proto "github.com/gogo/protobuf/proto"
+	"github.com/golang/snappy"
+	"github.com/prometheus/prometheus/prompb"
+)
+
+func TestTelemetryClient_Send(t *testing.T) {
+	clusterId := "fake-cluster-id"
+	pullSecret := "fake-pull-secret"
+	expectedBearer := fmt.Sprintf("Bearer %s", base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf(authString, pullSecret, clusterId))))
+	sampleMetrics := []Metric{
+		{
+			Name:      "fake-metric",
+			Labels:    []MetricLabel{{Name: "fake-label", Value: "fake-value"}},
+			Timestamp: 1234567890,
+			Value:     42,
+		},
+	}
+	expectedWriteRequest := prompb.WriteRequest{
+		Metadata: []prompb.MetricMetadata{
+			{MetricFamilyName: "fake-metric", Type: prompb.MetricMetadata_COUNTER},
+		},
+		Timeseries: []prompb.TimeSeries{
+			{
+				Labels: []prompb.Label{
+					{Name: "__name__", Value: "fake-metric"},
+					{Name: "fake-label", Value: "fake-value"},
+				},
+				Samples: []prompb.Sample{
+					{Timestamp: 1234567890, Value: 42},
+				},
+			},
+		},
+	}
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("Content-Type") != "application/x-protobuf" {
+			t.Fatalf("Expected Content-Type application/x-protobuf, got %s", r.Header.Get("Content-Type"))
+		}
+		if r.Header.Get("Content-Encoding") != "snappy" {
+			t.Fatalf("Expected Content-Encoding snappy, got %s", r.Header.Get("Content-Encoding"))
+		}
+		if r.Header.Get("Authorization") != expectedBearer {
+			t.Fatalf("Expected Authorization '%s', got %s", expectedBearer, r.Header.Get("Authorization"))
+		}
+
+		bodyBytes, err := io.ReadAll(r.Body)
+		if err != nil {
+			t.Fatalf("Failed to read request body: %v", err)
+		}
+		writeBytes, err := snappy.Decode(nil, bodyBytes)
+		if err != nil {
+			t.Fatalf("Failed to decode snappy body: %v", err)
+		}
+		var req prompb.WriteRequest
+		if err := proto.Unmarshal(writeBytes, &req); err != nil {
+			t.Fatalf("failed to unmarshal WriteRequest: %v", err)
+		}
+		if !reflect.DeepEqual(req, expectedWriteRequest) {
+			t.Fatalf("Expected WriteRequest %v, got %v", expectedWriteRequest, req)
+		}
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	client := NewTelemetryClient(server.URL, clusterId)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	err := client.Send(ctx, pullSecret, sampleMetrics)
+	if err != nil {
+		t.Fatalf("Send method failed: %v", err)
+	}
+}