Introducing pipedstat model

BUILD.bazel

@@ -62,6 +62,7 @@ genrule(
 # gazelle:exclude pkg/model/logblock.pb.validate.go
 # gazelle:exclude pkg/model/notificationevent.pb.validate.go
 # gazelle:exclude pkg/model/piped.pb.validate.go
+# gazelle:exclude pkg/model/piped_stat.pb.validate.go
 # gazelle:exclude pkg/model/piped_stats.pb.validate.go
 # gazelle:exclude pkg/model/planpreview.pb.validate.go
 # gazelle:exclude pkg/model/project.pb.validate.go

cmd/pipecd/ops.go

@@ -155,7 +155,7 @@ func (s *ops) run(ctx context.Context, t cli.Telemetry) error {
 	}
 
 	rd := redis.NewRedis(s.cacheAddress, "")
-	statCache := rediscache.NewTTLHashCache(rd, pipedStatTTL, defaultPipedStatHashKey)
+	statCache := rediscache.NewHashCache(rd, defaultPipedStatHashKey)
 	psb := pipedstatsbuilder.NewPipedStatsBuilder(statCache, t.Logger)
 
 	// Start running admin server.

cmd/pipecd/server.go

@@ -65,7 +65,6 @@ var (
 
 const (
 	defaultPipedStatHashKey = "HASHKEY:PIPED:STATS"
-	pipedStatTTL            = 2 * time.Minute
 )
 
 type httpHandler interface {
@@ -190,7 +189,7 @@ func (s *server) run(ctx context.Context, t cli.Telemetry) error {
 	cmds := commandstore.NewStore(ds, cache, t.Logger)
 	is := insightstore.NewStore(fs)
 	cmdOutputStore := commandoutputstore.NewStore(fs, t.Logger)
-	statCache := rediscache.NewTTLHashCache(rd, pipedStatTTL, defaultPipedStatHashKey)
+	statCache := rediscache.NewHashCache(rd, defaultPipedStatHashKey)
 
 	// Start a gRPC server for handling PipedAPI requests.
 	{

pkg/app/api/grpcapi/piped_api.go

@@ -16,6 +16,7 @@ package grpcapi
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"time"
 
@@ -98,7 +99,18 @@ func (a *PipedAPI) ReportStat(ctx context.Context, req *pipedservice.ReportStatR
 	if err != nil {
 		return nil, err
 	}
-	if err := a.pipedStatCache.Put(pipedID, req.PipedStats); err != nil {
+
+	now := time.Now().Unix()
+	val, err := json.Marshal(model.PipedStat{PipedId: pipedID, Metrics: req.PipedStats, Timestamp: now})
+	if err != nil {
+		a.logger.Error("failed to store the reported piped stat",
+			zap.String("piped-id", pipedID),
+			zap.Error(err),
+		)
+		return nil, status.Error(codes.Internal, "failed to encode the reported piped stat")
+	}
+
+	if err := a.pipedStatCache.Put(pipedID, val); err != nil {
 		a.logger.Error("failed to store the reported piped stat",
 			zap.String("piped-id", pipedID),
 			zap.Error(err),

pkg/app/ops/pipedstatsbuilder/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/cache:go_default_library",
+        "//pkg/model:go_default_library",
         "@org_uber_go_zap//:go_default_library",
     ],
 )
@@ -19,6 +20,7 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//pkg/cache:go_default_library",
+        "//pkg/model: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

@@ -16,12 +16,14 @@ package pipedstatsbuilder
 
 import (
 	"bytes"
+	"encoding/json"
 	"errors"
 	"io"
 
 	"go.uber.org/zap"
 
 	"github.com/pipe-cd/pipe/pkg/cache"
+	"github.com/pipe-cd/pipe/pkg/model"
 )
 
 type PipedStatsBuilder struct {
@@ -47,10 +49,16 @@ func (b *PipedStatsBuilder) Build() (io.Reader, error) {
 		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))
+			b.logger.Error("failed to unmarshal piped stat data", zap.Error(err))
 			return nil, err
 		}
-		data = append(data, value)
+		ps := model.PipedStat{}
+		if err = json.Unmarshal(value, &ps); err != nil {
+			b.logger.Error("failed to unmarshal piped stat data", zap.Error(err))
+			return nil, err
+		}
+		// TODO: Filter returning piped metrics by value timestamp.
+		data = append(data, ps.Metrics)
 	}
 	return bytes.NewReader(bytes.Join(data, []byte("\n"))), nil
 }

pkg/app/ops/pipedstatsbuilder/builder_test.go

@@ -15,16 +15,19 @@
 package pipedstatsbuilder
 
 import (
+	"encoding/json"
 	"io"
 	"os"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 
 	"github.com/pipe-cd/pipe/pkg/cache"
+	"github.com/pipe-cd/pipe/pkg/model"
 )
 
 type mockBuilderBackend struct {
@@ -48,7 +51,8 @@ func (m *mockBuilderBackend) GetAll() (map[string]interface{}, error) {
 		if err != nil {
 			return nil, err
 		}
-		out[file] = data
+		val, _ := json.Marshal(model.PipedStat{Metrics: data, Timestamp: time.Now().Unix()})
+		out[file] = val
 	}
 	return out, nil
 }

pkg/cache/rediscache/hashcache.go

@@ -16,7 +16,6 @@ package rediscache
 
 import (
 	"errors"
-	"time"
 
 	redigo "github.com/gomodule/redigo/redis"
 
@@ -37,14 +36,6 @@ func NewHashCache(redis redis.Redis, key string) *RedisHashCache {
 	}
 }
 
-func NewTTLHashCache(redis redis.Redis, ttl time.Duration, key string) *RedisHashCache {
-	return &RedisHashCache{
-		redis: redis,
-		ttl:   uint(ttl.Seconds()),
-		key:   key,
-	}
-}
-
 func (r *RedisHashCache) Get(k string) (interface{}, error) {
 	conn := r.redis.Get()
 	defer conn.Close()

pkg/datastore/pipedstatsstore.go

@@ -27,6 +27,7 @@ var pipedStatsFactory = func() interface{} {
 	return &model.PipedStats{}
 }
 
+// Deprecated: PipedStats model is deprecated, along with its store interface.
 type PipedStatsStore interface {
 	AddPipedStats(ctx context.Context, ps *model.PipedStats) error
 	ListPipedStatss(ctx context.Context, opts ListOptions) ([]model.PipedStats, error)
@@ -37,6 +38,7 @@ type pipedStatsStore struct {
 	nowFunc func() time.Time
 }
 
+// Deprecated: PipedStats model is deprecated, along with its store interface.
 func NewPipedStatsStore(ds DataStore) PipedStatsStore {
 	return &pipedStatsStore{
 		backend: backend{

pkg/model/BUILD.bazel

@@ -16,6 +16,7 @@ proto_library(
         "logblock.proto",
         "notificationevent.proto",
         "piped.proto",
+        "piped_stat.proto",
         "piped_stats.proto",
         "planpreview.proto",
         "project.proto",

pkg/model/piped_stat.proto

@@ -0,0 +1,26 @@
+// 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.
+
+syntax = "proto3";
+
+package pipe.model;
+option go_package = "github.com/pipe-cd/pipe/pkg/model";
+
+import "validate/validate.proto";
+
+message PipedStat {
+    string piped_id = 1 [(validate.rules).string.min_len = 1];
+    bytes metrics = 2;
+    int64 timestamp = 10 [(validate.rules).int64.gt = 0];
+}

pkg/model/piped_stats.proto

@@ -39,6 +39,7 @@ message PrometheusMetrics {
 }
 
 message PipedStats {
+    option deprecated = true;
     string version = 1 [(validate.rules).string.min_len = 1];
     int64 timestamp = 2 [(validate.rules).int64.gt = 0];
     repeated PrometheusMetrics prometheus_metrics = 3;