Implement live state reporter for Cloud Run

pkg/app/piped/livestatereporter/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     importpath = "github.com/pipe-cd/pipecd/pkg/app/piped/livestatereporter",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/app/piped/livestatereporter/cloudrun:go_default_library",
         "//pkg/app/piped/livestatereporter/kubernetes:go_default_library",
         "//pkg/app/piped/livestatestore:go_default_library",
         "//pkg/app/server/service/pipedservice:go_default_library",

pkg/app/piped/livestatereporter/cloudrun/BUILD.bazel

@@ -0,0 +1,16 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["report.go"],
+    importpath = "github.com/pipe-cd/pipecd/pkg/app/piped/livestatereporter/cloudrun",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/app/piped/livestatestore/cloudrun:go_default_library",
+        "//pkg/app/server/service/pipedservice:go_default_library",
+        "//pkg/config:go_default_library",
+        "//pkg/model:go_default_library",
+        "@org_golang_google_grpc//:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)

pkg/app/piped/livestatereporter/cloudrun/report.go

@@ -0,0 +1,134 @@
+// Copyright 2022 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 cloudrun
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"go.uber.org/zap"
+	"google.golang.org/grpc"
+
+	"github.com/pipe-cd/pipecd/pkg/app/piped/livestatestore/cloudrun"
+	"github.com/pipe-cd/pipecd/pkg/app/server/service/pipedservice"
+	"github.com/pipe-cd/pipecd/pkg/config"
+	"github.com/pipe-cd/pipecd/pkg/model"
+)
+
+type applicationLister interface {
+	ListByCloudProvider(name string) []*model.Application
+}
+
+type apiClient interface {
+	ReportApplicationLiveState(ctx context.Context, req *pipedservice.ReportApplicationLiveStateRequest, opts ...grpc.CallOption) (*pipedservice.ReportApplicationLiveStateResponse, error)
+	ReportApplicationLiveStateEvents(ctx context.Context, req *pipedservice.ReportApplicationLiveStateEventsRequest, opts ...grpc.CallOption) (*pipedservice.ReportApplicationLiveStateEventsResponse, error)
+}
+
+type Reporter interface {
+	Run(ctx context.Context) error
+	ProviderName() string
+}
+
+type reporter struct {
+	provider              config.PipedCloudProvider
+	appLister             applicationLister
+	stateGetter           cloudrun.Getter
+	apiClient             apiClient
+	snapshotFlushInterval time.Duration
+	logger                *zap.Logger
+
+	snapshotVersions map[string]model.ApplicationLiveStateVersion
+}
+
+func NewReporter(cp config.PipedCloudProvider, appLister applicationLister, stateGetter cloudrun.Getter, apiClient apiClient, logger *zap.Logger) Reporter {
+	logger = logger.Named("cloudrun-reporter").With(
+		zap.String("cloud-provider", cp.Name),
+	)
+	return &reporter{
+		provider:              cp,
+		appLister:             appLister,
+		stateGetter:           stateGetter,
+		apiClient:             apiClient,
+		snapshotFlushInterval: 10 * time.Minute,
+		logger:                logger,
+		snapshotVersions:      make(map[string]model.ApplicationLiveStateVersion),
+	}
+}
+
+func (r *reporter) Run(ctx context.Context) error {
+	r.logger.Info("start running app live state reporter")
+
+	r.logger.Info("waiting for livestatestore to be ready")
+	if err := r.stateGetter.WaitForReady(ctx, 10*time.Minute); err != nil {
+		r.logger.Error("livestatestore was unable to be ready in time", zap.Error(err))
+		return err
+	}
+
+	snapshotTicker := time.NewTicker(r.snapshotFlushInterval)
+	defer snapshotTicker.Stop()
+
+	for {
+		select {
+		case <-snapshotTicker.C:
+			r.flushSnapshots(ctx)
+
+		case <-ctx.Done():
+			r.logger.Info("app live state reporter has been stopped")
+			return nil
+		}
+	}
+}
+
+func (r *reporter) ProviderName() string {
+	return r.provider.Name
+}
+
+func (r *reporter) flushSnapshots(ctx context.Context) error {
+	apps := r.appLister.ListByCloudProvider(r.provider.Name)
+	for _, app := range apps {
+		state, ok := r.stateGetter.GetState(app.Id)
+		if !ok {
+			r.logger.Info(fmt.Sprintf("no app state of cloudrun application %s to report", app.Id))
+			continue
+		}
+
+		snapshot := &model.ApplicationLiveStateSnapshot{
+			ApplicationId: app.Id,
+			PipedId:       app.PipedId,
+			ProjectId:     app.ProjectId,
+			Kind:          app.Kind,
+			Cloudrun: &model.CloudRunApplicationLiveState{
+				Resources: state.Resources,
+			},
+			Version: &state.Version,
+		}
+		snapshot.DetermineAppHealthStatus()
+		req := &pipedservice.ReportApplicationLiveStateRequest{
+			Snapshot: snapshot,
+		}
+
+		if _, err := r.apiClient.ReportApplicationLiveState(ctx, req); err != nil {
+			r.logger.Error("failed to report application live state",
+				zap.String("application-id", app.Id),
+				zap.Error(err),
+			)
+			continue
+		}
+		r.snapshotVersions[app.Id] = state.Version
+		r.logger.Info(fmt.Sprintf("successfully reported application live state for application: %s", app.Id))
+	}
+	return nil
+}

pkg/app/piped/livestatereporter/reporter.go

@@ -25,6 +25,7 @@ import (
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 
+	"github.com/pipe-cd/pipecd/pkg/app/piped/livestatereporter/cloudrun"
 	"github.com/pipe-cd/pipecd/pkg/app/piped/livestatereporter/kubernetes"
 	"github.com/pipe-cd/pipecd/pkg/app/piped/livestatestore"
 	"github.com/pipe-cd/pipecd/pkg/app/server/service/pipedservice"
@@ -62,16 +63,22 @@ func NewReporter(appLister applicationLister, stateGetter livestatestore.Getter,
 	}
 
 	for _, cp := range cfg.CloudProviders {
+		errFmt := fmt.Sprintf("unable to find live state getter for cloud provider: %s", cp.Name)
 		switch cp.Type {
 		case model.CloudProviderKubernetes:
 			sg, ok := stateGetter.KubernetesGetter(cp.Name)
 			if !ok {
-				r.logger.Error(fmt.Sprintf("unable to find live state getter for cloud provider: %s", cp.Name))
+				r.logger.Error(errFmt)
 				continue
 			}
 			r.reporters = append(r.reporters, kubernetes.NewReporter(cp, appLister, sg, apiClient, logger))
-
-		default:
+		case model.CloudProviderCloudRun:
+			sg, ok := stateGetter.CloudRunGetter(cp.Name)
+			if !ok {
+				r.logger.Error(errFmt)
+				continue
+			}
+			r.reporters = append(r.reporters, cloudrun.NewReporter(cp, appLister, sg, apiClient, logger))
 		}
 	}