chore: restore firelog exporter (#8555)

pkg/skaffold/instrumentation/ci.go

@@ -0,0 +1,32 @@
+/*
+Copyright 2020 The Skaffold Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package instrumentation
+
+var ciMap = map[string]string{
+	"TF_BUILD":           "azure-pipelines",
+	"bamboo_buildKey":    "bamboo",
+	"BUILDKITE":          "buildkite",
+	"CIRCLECI":           "circle-ci",
+	"CIRRUS_CI":          "cirrus-ci",
+	"CODEBUILD_BUILD_ID": "code-build",
+	"GITHUB_ACTIONS":     "github-actions",
+	"GITLAB_CI":          "gitlab-ci",
+	"HEROKU_TEST_RUN_ID": "heroku-ci",
+	"JENKINS_URL":        "jenkins",
+	"TEAMCITY_VERSION":   "team-city",
+	"TRAVIS":             "travis-ci",
+}

pkg/skaffold/instrumentation/export.go

@@ -94,6 +94,11 @@ func exportMetrics(ctx context.Context, filename string, meter skaffoldMeter) er
 	}
 	var meters []skaffoldMeter
 	err = json.Unmarshal(b, &meters)
+	// each meter contains around 20 datapoints, and each datapoint requires a request to firelog api
+	// we send at most 10 meters stored in skaffold metrics as too many request may result in the firelog server returning 429.
+	if len(meters) >= 10 {
+		meters = meters[:10]
+	}
 	if err != nil {
 		meters = []skaffoldMeter{}
 	}
@@ -160,17 +165,13 @@ func devStdOutExporter() (sdkmetric.Exporter, error) {
 }
 
 func createMetrics(ctx context.Context, meter skaffoldMeter) {
-	// There is a minimum 10 second interval that metrics are allowed to upload to Cloud monitoring
+	m := global.Meter("skaffold")
+
 	// A metric is uniquely identified by the metric name and the labels and corresponding values
 	// This random number is used as a label to differentiate the metrics per user so if two users
 	// run `skaffold build` at the same time they will both have their metrics recorded
 	randLabel := attribute.String("randomizer", strconv.Itoa(rand.Intn(75000)))
 
-	m := global.Meter("skaffold")
-
-	// cloud monitoring only supports string type labels
-	// cloud monitoring only supports 10 labels per metric descriptor
-	// be careful when appending new values to this `labels` slice
 	labels := []attribute.KeyValue{
 		attribute.String("version", meter.Version),
 		attribute.String("os", meter.OS),
@@ -180,15 +181,13 @@ func createMetrics(ctx context.Context, meter skaffoldMeter) {
 		attribute.String("platform_type", meter.PlatformType),
 		attribute.String("config_count", strconv.Itoa(meter.ConfigCount)),
 		attribute.String("cluster_type", meter.ClusterType),
-	}
-	sharedLabels := []attribute.KeyValue{
+		attribute.String("ci_cd_platform", meter.CISystem),
 		randLabel,
 	}
 
 	if allowedUser := user.IsAllowedUser(meter.User); allowedUser {
-		sharedLabels = append(sharedLabels, attribute.String("user", meter.User))
+		labels = append(labels, attribute.String("user", meter.User))
 	}
-	labels = append(labels, sharedLabels...)
 	platformLabel := attribute.String("host_os_arch", fmt.Sprintf("%s/%s", meter.OS, meter.Arch))
 	runCounter := NewInt64ValueRecorder(m, "launches", instrument.WithDescription("Skaffold Invocations"))
 	runCounter.Record(ctx, 1, labels...)
@@ -197,36 +196,36 @@ func createMetrics(ctx context.Context, meter skaffoldMeter) {
 		instrument.WithDescription("durations of skaffold commands in seconds"))
 	durationRecorder.Record(ctx, meter.Duration.Seconds(), labels...)
 	if meter.Command != "" {
-		commandMetrics(ctx, meter, m, sharedLabels...)
-		flagMetrics(ctx, meter, m, randLabel)
+		commandMetrics(ctx, meter, m, labels...)
+		flagMetrics(ctx, meter, m, labels...)
 		hooksMetrics(ctx, meter, m, labels...)
 		if doesBuild.Contains(meter.Command) {
-			builderMetrics(ctx, meter, m, platformLabel, sharedLabels...)
+			builderMetrics(ctx, meter, m, platformLabel, labels...)
 		}
 		if doesDeploy.Contains(meter.Command) {
-			deployerMetrics(ctx, meter, m, sharedLabels...)
+			deployerMetrics(ctx, meter, m, labels...)
 		}
 		if doesDeploy.Contains(meter.Command) || meter.Command == "render" {
-			resourceSelectorMetrics(ctx, meter, m, sharedLabels...)
+			resourceSelectorMetrics(ctx, meter, m, labels...)
 		}
 	}
 
 	if meter.ErrorCode != 0 {
-		errorMetrics(ctx, meter, m, append(sharedLabels, platformLabel)...)
+		errorMetrics(ctx, meter, m, append(labels, platformLabel)...)
 	}
 }
 
-func flagMetrics(ctx context.Context, meter skaffoldMeter, m metric.Meter, randLabel attribute.KeyValue) {
+func flagMetrics(ctx context.Context, meter skaffoldMeter, m metric.Meter, labels ...attribute.KeyValue) {
 	flagCounter := NewInt64ValueRecorder(m, "flags", instrument.WithDescription("Tracks usage of enum flags"))
 	for k, v := range meter.EnumFlags {
-		labels := []attribute.KeyValue{
+		l := []attribute.KeyValue{
 			attribute.String("flag_name", k),
 			attribute.String("flag_value", v),
 			attribute.String("command", meter.Command),
 			attribute.String("error", meter.ErrorCode.String()),
-			randLabel,
 		}
-		flagCounter.Record(ctx, 1, labels...)
+		l = append(l, labels...)
+		flagCounter.Record(ctx, 1, l...)
 	}
 }
 
@@ -249,12 +248,13 @@ func commandMetrics(ctx context.Context, meter skaffoldMeter, m metric.Meter, la
 			m := counts[iteration.Intent]
 			m[iteration.ErrorCode]++
 		}
+		randomizer := attribute.String("randomizer2", strconv.Itoa(rand.Intn(75000)))
 		for intention, errorCounts := range counts {
 			for errorCode, count := range errorCounts {
 				iterationCounter.Record(ctx, int64(count),
 					append(labels,
 						attribute.String("intent", intention),
-						attribute.String("error", errorCode.String()),
+						attribute.String("error", errorCode.String()), randomizer,
 					)...)
 			}
 		}

pkg/skaffold/instrumentation/export_test.go

@@ -31,6 +31,7 @@ import (
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
 
 	"github.com/GoogleContainerTools/skaffold/v2/fs"
+	"github.com/GoogleContainerTools/skaffold/v2/pkg/skaffold/instrumentation/firelog"
 	"github.com/GoogleContainerTools/skaffold/v2/proto/v1"
 	"github.com/GoogleContainerTools/skaffold/v2/testutil"
 )
@@ -105,7 +106,7 @@ func TestExportMetrics(t *testing.T) {
 		StartTime:                    startTime.Add(time.Hour * 24 * 10),
 		Duration:                     time.Minute * 4,
 	}
-	metersBytes, _ := json.Marshal([]skaffoldMeter{buildMeter, devMeter, debugMeter})
+	metersBytes, _ := json.Marshal([]skaffoldMeter{devMeter, debugMeter, buildMeter})
 	fakeFS := testutil.FakeFileSystem{
 		Files: map[string][]byte{
 			"assets/secrets_generated/keys.json": []byte(testKey),
@@ -168,6 +169,7 @@ func TestExportMetrics(t *testing.T) {
 
 			fs.AssetsFS = fakeFS
 			t.Override(&isOnline, test.isOnline)
+			t.Override(&firelog.APIKey, "no-empty")
 
 			if test.isOnline {
 				tmpFile, err := os.OpenFile(tmp.Path(openTelFilename), os.O_RDWR|os.O_CREATE, os.ModePerm)
@@ -406,6 +408,7 @@ func checkOutput(t *testutil.T, meters []skaffoldMeter, b []byte) {
 	platform := make(map[interface{}]int)
 	buildPlatforms := make(map[interface{}]int)
 	cliPlatforms := make(map[interface{}]int)
+	ciCDPlatforms := make(map[interface{}]int)
 	nodePlatforms := make(map[interface{}]int)
 
 	testMaps := []map[interface{}]int{
@@ -419,6 +422,7 @@ func checkOutput(t *testutil.T, meters []skaffoldMeter, b []byte) {
 		commandCount[meter.Command]++
 		errorCount[meter.ErrorCode.String()]++
 		platform[meter.PlatformType]++
+		ciCDPlatforms[meter.CISystem]++
 
 		for k, v := range meter.EnumFlags {
 			n := strings.ReplaceAll(k, "-", "_")
@@ -485,10 +489,6 @@ func checkOutput(t *testutil.T, meters []skaffoldMeter, b []byte) {
 			}
 		}
 	}
-	for _, l := range lines {
-		fmt.Println(l.Name)
-		fmt.Println(l.Labels)
-	}
 
 	for _, l := range lines {
 		switch l.Name {
@@ -578,6 +578,10 @@ func checkOutput(t *testutil.T, meters []skaffoldMeter, b []byte) {
 			if v, ok := l.Labels["cli-platforms"]; ok {
 				cliPlatforms[v]++
 			}
+		case "ci-cd-platforms":
+			if v, ok := l.Labels["ci-cd-platforms"]; ok {
+				ciCDPlatforms[v]++
+			}
 		default:
 			switch {
 			case MeteredCommands.Contains(l.Name):

pkg/skaffold/instrumentation/firelog/exporter.go

@@ -0,0 +1,185 @@
+/*
+Copyright 2020 The Skaffold Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package firelog
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"os"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/exporters/stdout/stdoutmetric"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+)
+
+var (
+	APIKey  = ""
+	url     = fmt.Sprintf(`https://firebaselogging-pa.googleapis.com/v1/firelog/legacy/log?key=%s`, APIKey)
+	POST    = http.Post
+	Marshal = json.Marshal
+)
+
+type Exporter struct {
+}
+
+func NewFireLogExporter() (metric.Exporter, error) {
+	if APIKey == "" {
+		// export metrics to std out if local env is set.
+		if _, ok := os.LookupEnv("SKAFFOLD_EXPORT_TO_STDOUT"); ok {
+			enc := json.NewEncoder(os.Stdout)
+			enc.SetIndent("", "  ")
+			exporter, err := stdoutmetric.New(stdoutmetric.WithEncoder(enc))
+			return exporter, err
+		}
+		return nil, nil
+	}
+	return &Exporter{}, nil
+}
+
+// Temporality returns the Temporality to use for an instrument kind.
+func (e *Exporter) Temporality(ik metric.InstrumentKind) metricdata.Temporality {
+	return metric.DefaultTemporalitySelector(ik)
+}
+
+// Aggregation returns the Aggregation to use for an instrument kind.
+func (e *Exporter) Aggregation(ik metric.InstrumentKind) aggregation.Aggregation {
+	return metric.DefaultAggregationSelector(ik)
+}
+
+func (e *Exporter) Export(ctx context.Context, md metricdata.ResourceMetrics) error {
+	for _, sm := range md.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			if err := processMetrics(m); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func processMetrics(m metricdata.Metrics) error {
+	switch a := m.Data.(type) {
+	case metricdata.Gauge[int64]:
+		for _, pt := range a.DataPoints {
+			if err := sendDataPoint(m.Name, DataPointInt64(pt)); err != nil {
+				return err
+			}
+		}
+	case metricdata.Gauge[float64]:
+		for _, pt := range a.DataPoints {
+			if err := sendDataPoint(m.Name, DataPointFloat64(pt)); err != nil {
+				return err
+			}
+		}
+	case metricdata.Sum[int64]:
+		for _, pt := range a.DataPoints {
+			if err := sendDataPoint(m.Name, DataPointInt64(pt)); err != nil {
+				return err
+			}
+		}
+	case metricdata.Sum[float64]:
+		for _, pt := range a.DataPoints {
+			if err := sendDataPoint(m.Name, DataPointFloat64(pt)); err != nil {
+				return err
+			}
+		}
+	case metricdata.Histogram:
+		for _, pt := range a.DataPoints {
+			if err := sendDataPoint(m.Name, DataPointHistogram(pt)); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func sendDataPoint(name string, dp DataPoint) error {
+	kvs := toEventMetadata(dp.attributes())
+	kvs = append(kvs, KeyValue{Key: name, Value: dp.value()})
+	str, err := buildProtoStr(name, kvs)
+	if err != nil {
+		return err
+	}
+	data := buildMetricData(str, dp.eventTime(), dp.upTime())
+
+	resp, err := POST(url, "application/json", data.newReader())
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("one platform returned an non-success response: %d", resp.StatusCode)
+	}
+
+	return err
+}
+
+func buildMetricData(proto string, startTimeMS int64, upTimeMS int64) MetricData {
+	return MetricData{
+		ClientInfo: ClientInfo{ClientType: "DESKTOP"},
+		LogSource:  "CONCORD",
+		LogEvent: LogEvent{
+			EventTimeMS:                  startTimeMS,
+			EventUptimeMS:                upTimeMS,
+			SourceExtensionJSONProto3Str: proto,
+		},
+		RequestTimeMS:   startTimeMS,
+		RequestUptimeMS: upTimeMS,
+	}
+}
+
+func buildProtoStr(name string, kvs EventMetadata) (string, error) {
+	proto3 := SourceExtensionJSONProto3{
+		ProjectID:       "skaffold",
+		ConsoleType:     "SKAFFOLD",
+		ClientInstallID: "",
+		EventName:       name,
+		EventMetadata:   kvs,
+	}
+
+	b, err := Marshal(proto3)
+	if err != nil {
+		return "", fmt.Errorf("failed to marshal metricdata")
+	}
+	return string(b), nil
+}
+
+func toEventMetadata(attributes attribute.Set) EventMetadata {
+	kvs := EventMetadata{}
+	iterator := attributes.Iter()
+	for iterator.Next() {
+		attr := iterator.Attribute()
+		kv := KeyValue{
+			Key:   string(attr.Key),
+			Value: attr.Value.Emit(),
+		}
+		kvs = append(kvs, kv)
+	}
+	return kvs
+}
+
+func (e *Exporter) ForceFlush(ctx context.Context) error {
+	return ctx.Err()
+}
+
+func (e *Exporter) Shutdown(ctx context.Context) error {
+	return ctx.Err()
+}