Add a piped component that watches app configs

pkg/app/api/grpcapi/piped_api.go

@@ -933,6 +933,16 @@ func (a *PipedAPI) GetDesiredVersion(ctx context.Context, _ *pipedservice.GetDes
 	}, nil
 }
 
+func (a *PipedAPI) UpdateApplicationConfigurations(ctx context.Context, req *pipedservice.UpdateApplicationConfigurationsRequest) (*pipedservice.UpdateApplicationConfigurationsResponse, error) {
+	// TODO: Update the given application configurations
+	return nil, status.Errorf(codes.Unimplemented, "UpdateApplicationConfigurations is not implemented yet")
+}
+
+func (a *PipedAPI) ReportUnregisteredApplicationConfigurations(ctx context.Context, req *pipedservice.ReportUnregisteredApplicationConfigurationsRequest) (*pipedservice.ReportUnregisteredApplicationConfigurationsResponse, error) {
+	// TODO: Make the unused application configurations cache up-to-date
+	return nil, status.Errorf(codes.Unimplemented, "ReportUnregisteredApplicationConfigurations is not implemented yet")
+}
+
 // validateAppBelongsToPiped checks if the given application belongs to the given piped.
 // It gives back an error unless the application belongs to the piped.
 func (a *PipedAPI) validateAppBelongsToPiped(ctx context.Context, appID, pipedID string) error {

pkg/app/api/service/pipedservice/service.proto

@@ -162,7 +162,10 @@ service PipedService {
     // GetDesiredVersion returns the desired version of the given Piped.
     rpc GetDesiredVersion(GetDesiredVersionRequest) returns (GetDesiredVersionResponse) {}
 
-    // TODO: Add an rpc to update application info based on one defined in the application config
+    // UpdateApplicationConfigurations updates application configurations.
+    rpc UpdateApplicationConfigurations(UpdateApplicationConfigurationsRequest) returns (UpdateApplicationConfigurationsResponse) {}
+    // ReportLatestUnusedApplicationConfigurations puts the latest configurations of applications that isn't registered yet.
+    rpc ReportUnregisteredApplicationConfigurations(ReportUnregisteredApplicationConfigurationsRequest) returns (ReportUnregisteredApplicationConfigurationsResponse) {}
 }
 
 enum ListOrder {
@@ -445,3 +448,21 @@ message GetDesiredVersionRequest {
 message GetDesiredVersionResponse {
     string version = 1;
 }
+
+
+message UpdateApplicationConfigurationsRequest {
+    // The application configurations that should be updated.
+    repeated pipe.model.ApplicationInfo applications = 1;
+}
+
+message UpdateApplicationConfigurationsResponse {
+}
+
+message ReportUnregisteredApplicationConfigurationsRequest {
+    // All the latest application configurations that isn't registered yet.
+    // Note that ALL configs are always be contained every time.
+    repeated pipe.model.ApplicationInfo applications = 1;
+}
+
+message ReportUnregisteredApplicationConfigurationsResponse {
+}

pkg/app/piped/apistore/environmentstore/store.go

@@ -34,27 +34,33 @@ const (
 
 type apiClient interface {
 	GetEnvironment(ctx context.Context, in *pipedservice.GetEnvironmentRequest, opts ...grpc.CallOption) (*pipedservice.GetEnvironmentResponse, error)
+	//GetEnvironmentByName(ctx context.Context, in *pipedservice.GetEnvironmentByNameRequest, opts ...grpc.CallOption) (*pipedservice.GetEnvironmentByNameResponse, error)
 }
 
 // Lister helps list and get Environment.
 // All objects returned here must be treated as read-only.
 type Lister interface {
 	Get(ctx context.Context, id string) (*model.Environment, error)
+	GetByName(ctx context.Context, name string) (*model.Environment, error)
 }
 
 type Store struct {
 	apiClient apiClient
-	cache     cache.Cache
-	callGroup *singleflight.Group
-	logger    *zap.Logger
+	// A goroutine-safe map from id to Environment.
+	cache cache.Cache
+	// A goroutine-safe map from name to Environment.
+	cacheByName cache.Cache
+	callGroup   *singleflight.Group
+	logger      *zap.Logger
 }
 
-func NewStore(apiClient apiClient, cache cache.Cache, logger *zap.Logger) *Store {
+func NewStore(apiClient apiClient, cache, cacheByName cache.Cache, logger *zap.Logger) *Store {
 	return &Store{
-		apiClient: apiClient,
-		cache:     cache,
-		callGroup: &singleflight.Group{},
-		logger:    logger.Named("environmentstore"),
+		apiClient:   apiClient,
+		cache:       cache,
+		cacheByName: cacheByName,
+		callGroup:   &singleflight.Group{},
+		logger:      logger.Named("environmentstore"),
 	}
 }
 
@@ -94,3 +100,45 @@ func (s *Store) Get(ctx context.Context, id string) (*model.Environment, error)
 	}
 	return data.(*model.Environment), nil
 }
+
+// TODO: Implement environmentstore.GetByName
+func (s *Store) GetByName(ctx context.Context, name string) (*model.Environment, error) {
+	/*
+		env, err := s.cacheByName.Get(name)
+		if err == nil {
+			return env.(*model.Environment), nil
+		}
+
+		// Ensure that timeout is configured.
+		ctx, cancel := context.WithTimeout(ctx, defaultAPITimeout)
+		defer cancel()
+
+		// Ensure that only one RPC call is executed for the given key at a time
+		// and the newest data is stored in the cache.
+		data, err, _ := s.callGroup.Do(name, func() (interface{}, error) {
+			req := &pipedservice.GetEnvironmentByNameRequest{
+				Name: name,
+			}
+			resp, err := s.apiClient.GetEnvironmentByName(ctx, req)
+			if err != nil {
+				s.logger.Warn("failed to get environment from control plane",
+					zap.String("name", name),
+					zap.Error(err),
+				)
+				return nil, fmt.Errorf("failed to get environment %s, %w", name, err)
+			}
+
+			if err := s.cacheByName.Put(id, resp.Environment); err != nil {
+				s.logger.Warn("failed to put environment to cache", zap.Error(err))
+			}
+			return resp.Environment, nil
+		})
+
+		if err != nil {
+			return nil, err
+		}
+		return data.(*model.Environment), nil
+	*/
+
+	return nil, fmt.Errorf("not implemented")
+}

pkg/app/piped/appconfigreporter/BUILD.bazel

@@ -0,0 +1,28 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["appconfigreporter.go"],
+    importpath = "github.com/pipe-cd/pipe/pkg/app/piped/appconfigreporter",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/app/api/service/pipedservice:go_default_library",
+        "//pkg/config:go_default_library",
+        "//pkg/git:go_default_library",
+        "//pkg/model:go_default_library",
+        "@org_golang_google_grpc//:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    size = "small",
+    srcs = ["appconfigreporter_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//pkg/model:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)