Create insight collector

cmd/pipecd/BUILD.bazel

@@ -21,6 +21,7 @@ go_library(
         "//pkg/app/api/service/webservice:go_default_library",
         "//pkg/app/api/stagelogstore:go_default_library",
         "//pkg/app/ops/handler:go_default_library",
+        "//pkg/app/ops/insightcollector:go_default_library",
         "//pkg/cache/rediscache:go_default_library",
         "//pkg/cli:go_default_library",
         "//pkg/config:go_default_library",
@@ -38,6 +39,7 @@ go_library(
         "//pkg/rpc:go_default_library",
         "//pkg/version:go_default_library",
         "@com_github_dgrijalva_jwt_go//:go_default_library",
+        "@com_github_robfig_cron_v3//:go_default_library",
         "@com_github_nytimes_gziphandler//:go_default_library",
         "@com_github_spf13_cobra//:go_default_library",
         "@org_golang_x_sync//errgroup:go_default_library",

cmd/pipecd/ops.go

@@ -19,22 +19,25 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/robfig/cron/v3"
 	"github.com/spf13/cobra"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/pipe-cd/pipe/pkg/admin"
 	"github.com/pipe-cd/pipe/pkg/app/ops/handler"
+	"github.com/pipe-cd/pipe/pkg/app/ops/insightcollector"
 	"github.com/pipe-cd/pipe/pkg/cli"
 	"github.com/pipe-cd/pipe/pkg/datastore"
 	"github.com/pipe-cd/pipe/pkg/version"
 )
 
 type ops struct {
-	httpPort    int
-	adminPort   int
-	gracePeriod time.Duration
-	configFile  string
+	httpPort               int
+	adminPort              int
+	gracePeriod            time.Duration
+	enableInsightCollector bool
+	configFile             string
 }
 
 func NewOpsCommand() *cobra.Command {
@@ -51,7 +54,7 @@ func NewOpsCommand() *cobra.Command {
 	cmd.Flags().IntVar(&s.httpPort, "http-port", s.httpPort, "The port number used to run http server.")
 	cmd.Flags().IntVar(&s.adminPort, "admin-port", s.adminPort, "The port number used to run a HTTP server for admin tasks such as metrics, healthz.")
 	cmd.Flags().DurationVar(&s.gracePeriod, "grace-period", s.gracePeriod, "How long to wait for graceful shutdown.")
-
+	cmd.Flags().BoolVar(&s.enableInsightCollector, "enableInsightCollector-insight-collector", s.enableInsightCollector, "Enable insight collector.")
 	cmd.Flags().StringVar(&s.configFile, "config-file", s.configFile, "The path to the configuration file.")
 	return cmd
 }
@@ -82,6 +85,41 @@ func (s *ops) run(ctx context.Context, t cli.Telemetry) error {
 		}
 	}()
 
+	fs, err := createFilestore(ctx, cfg, t.Logger)
+	if err != nil {
+		t.Logger.Error("failed to create filestore", zap.Error(err))
+		return err
+	}
+	defer func() {
+		if err := fs.Close(); err != nil {
+			t.Logger.Error("failed to close filestore client", zap.Error(err))
+		}
+	}()
+
+	// Starting a cron job for insight collector.
+	if s.enableInsightCollector {
+		collector := insightcollector.NewInsightCollector(ds, fs, t.Logger)
+		c := cron.New(cron.WithLocation(time.UTC))
+		_, err := c.AddFunc(cfg.InsightCollector.Schedule, func() {
+			start := time.Now()
+			if err := collector.CollectProjectsInsight(ctx); err != nil {
+				t.Logger.Error("failed to run the project insight collector", zap.Error(err))
+			} else {
+				t.Logger.Info("project insight collector successfully finished", zap.Duration("duration", time.Since(start)))
+			}
+
+			start = time.Now()
+			if err := collector.CollectApplicationInsight(ctx); err != nil {
+				t.Logger.Error("failed to run the application insight collector", zap.Error(err))
+			} else {
+				t.Logger.Info("application insight collector successfully finished", zap.Duration("duration", time.Since(start)))
+			}
+		})
+		if err != nil {
+			t.Logger.Error("failed to configure the insight collector", zap.Error(err))
+		}
+	}
+
 	// Start running HTTP server.
 	{
 		handler := handler.NewHandler(s.httpPort, datastore.NewProjectStore(ds), cfg.SharedSSOConfigs, s.gracePeriod, t.Logger)

go.mod

@@ -28,6 +28,7 @@ require (
 	github.com/prometheus/client_golang v1.6.0
 	github.com/prometheus/client_model v0.2.0
 	github.com/prometheus/common v0.9.1
+	github.com/robfig/cron/v3 v3.0.1 // indirect
 	github.com/spf13/cobra v0.0.5
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.5.1

go.sum

@@ -395,6 +395,9 @@ github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsT
 github.com/prometheus/procfs v0.0.3/go.mod h1:4A/X28fw3Fc593LaREMrKMqOKvUAntwMDaekg4FpcdQ=
 github.com/prometheus/procfs v0.0.11 h1:DhHlBtkHWPYi8O2y31JkK0TF+DGM+51OopZjH/Ia5qI=
 github.com/prometheus/procfs v0.0.11/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
+github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.1.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.2.2/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=

pkg/app/ops/insightcollector/BUILD.bazel

@@ -1,8 +1,32 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
     srcs = ["collector.go"],
     importpath = "github.com/pipe-cd/pipe/pkg/app/ops/insightcollector",
     visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/datastore:go_default_library",
+        "//pkg/insightstore:go_default_library",
+        "//pkg/model:go_default_library",
+        "//pkg/filestore:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    size = "small",
+    srcs = [
+        "collector_test.go"
+    ],
+    embed = [":go_default_library"],
+    deps = [
+        "//pkg/insightstore:go_default_library",
+        "//pkg/filestore/filestoretest:go_default_library",
+        "//pkg/datastore:go_default_library",
+        "//pkg/datastore/datastoretest:go_default_library",
+        "@com_github_golang_mock//gomock:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+    ],
 )