Expose piped stat metrics at ops metrics endpoint

cmd/pipecd/BUILD.bazel

@@ -26,6 +26,7 @@ go_library(
         "//pkg/app/ops/insightcollector:go_default_library",
         "//pkg/app/ops/mysqlensurer:go_default_library",
         "//pkg/app/ops/orphancommandcleaner:go_default_library",
+        "//pkg/app/ops/pipedstatsbuilder:go_default_library",
         "//pkg/cache/cachemetrics:go_default_library",
         "//pkg/cache/rediscache:go_default_library",
         "//pkg/cli:go_default_library",

cmd/pipecd/ops.go

@@ -29,11 +29,14 @@ import (
 	"github.com/pipe-cd/pipe/pkg/app/ops/insightcollector"
 	"github.com/pipe-cd/pipe/pkg/app/ops/mysqlensurer"
 	"github.com/pipe-cd/pipe/pkg/app/ops/orphancommandcleaner"
+	"github.com/pipe-cd/pipe/pkg/app/ops/pipedstatsbuilder"
+	"github.com/pipe-cd/pipe/pkg/cache/rediscache"
 	"github.com/pipe-cd/pipe/pkg/cli"
 	"github.com/pipe-cd/pipe/pkg/config"
 	"github.com/pipe-cd/pipe/pkg/datastore"
 	"github.com/pipe-cd/pipe/pkg/insight/insightstore"
 	"github.com/pipe-cd/pipe/pkg/model"
+	"github.com/pipe-cd/pipe/pkg/redis"
 	"github.com/pipe-cd/pipe/pkg/version"
 )
 
@@ -44,13 +47,15 @@ type ops struct {
 	enableInsightCollector bool
 	configFile             string
 	gcloudPath             string
+	cacheAddress           string
 }
 
 func NewOpsCommand() *cobra.Command {
 	s := &ops{
-		httpPort:    9082,
-		adminPort:   9085,
-		gracePeriod: 15 * time.Second,
+		httpPort:     9082,
+		adminPort:    9085,
+		cacheAddress: "cache:6379",
+		gracePeriod:  15 * time.Second,
 	}
 	cmd := &cobra.Command{
 		Use:   "ops",
@@ -62,6 +67,7 @@ func NewOpsCommand() *cobra.Command {
 	cmd.Flags().DurationVar(&s.gracePeriod, "grace-period", s.gracePeriod, "How long to wait for graceful shutdown.")
 	cmd.Flags().StringVar(&s.configFile, "config-file", s.configFile, "The path to the configuration file.")
 	cmd.Flags().StringVar(&s.gcloudPath, "gcloud-path", s.gcloudPath, "The path to the gcloud command executable.")
+	cmd.Flags().StringVar(&s.cacheAddress, "cache-address", s.cacheAddress, "The address to cache service.")
 	return cmd
 }
 
@@ -144,6 +150,10 @@ func (s *ops) run(ctx context.Context, t cli.Telemetry) error {
 		})
 	}
 
+	rd := redis.NewRedis(s.cacheAddress, "")
+	statCache := rediscache.NewTTLHashCache(rd, cfg.Cache.TTLDuration(), defaultPipedStatHashKey)
+	psb := pipedstatsbuilder.NewPipedStatsBuilder(statCache, t.Logger)
+
 	// Start running admin server.
 	{
 		var (
@@ -157,7 +167,7 @@ func (s *ops) run(ctx context.Context, t cli.Telemetry) error {
 		admin.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
 			w.Write([]byte("ok"))
 		})
-		admin.Handle("/metrics", t.PrometheusMetricsHandler())
+		admin.Handle("/metrics", t.CustomMetricsHandlerFor(psb))
 
 		group.Go(func() error {
 			return admin.Run(ctx)

manifests/pipecd/templates/deployment.yaml

@@ -215,6 +215,7 @@ spec:
           imagePullPolicy: IfNotPresent
           args:
           - ops
+          - --cache-address={{ .Values.ops.args.cacheAddress | default (printf "%s-cache:6379" (include "pipecd.fullname" .)) }}
           - --config-file=/etc/pipecd-config/{{ .Values.config.fileName }}
           - --log-encoding={{ .Values.ops.args.logEncoding }}
           ports:

manifests/pipecd/values.yaml

@@ -40,6 +40,7 @@ ops:
   image:
     repository: gcr.io/pipecd/pipecd
   args:
+    cacheAddress: ""
     logEncoding: humanize
   resources: {}
 

pkg/app/ops/pipedstatsbuilder/BUILD.bazel

@@ -0,0 +1,26 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["builder.go"],
+    importpath = "github.com/pipe-cd/pipe/pkg/app/ops/pipedstatsbuilder",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/cache:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    size = "small",
+    srcs = ["builder_test.go"],
+    data = glob(["testdata/**"]),
+    embed = [":go_default_library"],
+    deps = [
+        "//pkg/cache:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)

pkg/app/ops/pipedstatsbuilder/builder.go

@@ -0,0 +1,56 @@
+// Copyright 2021 The PipeCD 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 pipedstatsbuilder
+
+import (
+	"bytes"
+	"errors"
+	"io"
+
+	"go.uber.org/zap"
+
+	"github.com/pipe-cd/pipe/pkg/cache"
+)
+
+type PipedStatsBuilder struct {
+	backend cache.Cache
+	logger  *zap.Logger
+}
+
+func NewPipedStatsBuilder(c cache.Cache, logger *zap.Logger) *PipedStatsBuilder {
+	return &PipedStatsBuilder{
+		backend: c,
+		logger:  logger.Named("piped-metrics-builder"),
+	}
+}
+
+func (b *PipedStatsBuilder) Build() (io.Reader, error) {
+	res, err := b.backend.GetAll()
+	if err != nil {
+		b.logger.Error("failed to fetch piped stats from cache", zap.Error(err))
+		return nil, err
+	}
+	data := make([][]byte, 0, len(res))
+	for _, v := range res {
+		value, okValue := v.([]byte)
+		if !okValue {
+			err = errors.New("error value not a bulk of string value")
+			b.logger.Error("failed to marshal piped stat data", zap.Error(err))
+			return nil, err
+		}
+		data = append(data, value)
+	}
+	return bytes.NewReader(bytes.Join(data, []byte("\n"))), nil
+}

pkg/app/ops/pipedstatsbuilder/builder_test.go

@@ -0,0 +1,65 @@
+// Copyright 2021 The PipeCD 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 pipedstatsbuilder
+
+import (
+	"io"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+
+	"github.com/pipe-cd/pipe/pkg/cache"
+)
+
+type mockBuilderBackend struct {
+	cache.Cache
+	srcs []string
+}
+
+func newMockBuilderBackend() *mockBuilderBackend {
+	return &mockBuilderBackend{
+		srcs: []string{
+			"./testdata/piped_stat_1",
+			"./testdata/piped_stat_2",
+		},
+	}
+}
+
+func (m *mockBuilderBackend) GetAll() (map[string]interface{}, error) {
+	out := make(map[string]interface{}, len(m.srcs))
+	for _, file := range m.srcs {
+		data, err := os.ReadFile(file)
+		if err != nil {
+			return nil, err
+		}
+		out[file] = data
+	}
+	return out, nil
+}
+
+func TestBuildPipedStat(t *testing.T) {
+	builder := NewPipedStatsBuilder(newMockBuilderBackend(), zap.NewNop())
+	rc, err := builder.Build()
+	require.NoError(t, err)
+	require.NotNil(t, rc)
+	buf := new(strings.Builder)
+	io.Copy(buf, rc)
+	data, _ := os.ReadFile("./testdata/expected")
+	assert.Equal(t, string(data), buf.String())
+}