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.CustomedMetricsHandlerFor(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,12 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+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",
+    ],
+)

pkg/app/ops/pipedstatsbuilder/builder.go

@@ -0,0 +1,57 @@
+// 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"
+	"io/ioutil"
+
+	"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.ReadCloser, 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 ioutil.NopCloser(bytes.NewReader(bytes.Join(data, []byte("\n")))), nil
+}

pkg/cli/cmd.go

@@ -16,6 +16,7 @@ package cli
 
 import (
 	"context"
+	"io"
 	"net/http"
 	"os"
 	"os/signal"
@@ -137,6 +138,31 @@ func (t Telemetry) PrometheusMetricsHandlerFor(r *prometheus.Registry) http.Hand
 	return empty
 }
 
+type MetricsBuilder interface {
+	Build() (io.ReadCloser, error)
+}
+
+func (t Telemetry) CustomedMetricsHandlerFor(mb MetricsBuilder) http.Handler {
+	if t.Flags.Metrics {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			rc, err := mb.Build()
+			if err != nil {
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+			}
+			defer rc.Close()
+
+			_, err = io.Copy(w, rc)
+			if err != nil {
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+			}
+		})
+	}
+	var empty http.HandlerFunc = func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte(""))
+	}
+	return empty
+}
+
 func extractServiceName(cmd *cobra.Command) string {
 	return strings.Replace(cmd.CommandPath(), " ", ".", -1)
 }