Send remaining notification events before exiting and fix env name

pkg/app/piped/apistore/environmentstore/BUILD.bazel

@@ -0,0 +1,22 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["store.go"],
+    importpath = "github.com/pipe-cd/pipe/pkg/app/piped/apistore/environmentstore",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/app/api/service/pipedservice:go_default_library",
+        "//pkg/cache: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 = ["store_test.go"],
+    embed = [":go_default_library"],
+)

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

@@ -0,0 +1,80 @@
+// Copyright 2020 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 environmentstore
+
+import (
+	"context"
+	"time"
+
+	"go.uber.org/zap"
+	"google.golang.org/grpc"
+
+	"github.com/pipe-cd/pipe/pkg/app/api/service/pipedservice"
+	"github.com/pipe-cd/pipe/pkg/cache"
+	"github.com/pipe-cd/pipe/pkg/model"
+)
+
+type apiClient interface {
+	GetEnvironment(ctx context.Context, in *pipedservice.GetEnvironmentRequest, opts ...grpc.CallOption) (*pipedservice.GetEnvironmentResponse, error)
+}
+
+// Lister helps list and get application.
+// All objects returned here must be treated as read-only.
+type Lister interface {
+	// Get retrieves a specifiec Environment for the given id.
+	Get(id string) (*model.Environment, bool)
+}
+
+type Store struct {
+	apiClient  apiClient
+	cache      cache.Cache
+	apiTimeout time.Duration
+	logger     *zap.Logger
+}
+
+func NewStore(apiClient apiClient, cache cache.Cache, logger *zap.Logger) *Store {
+	return &Store{
+		apiClient:  apiClient,
+		cache:      cache,
+		apiTimeout: 10 * time.Second,
+		logger:     logger.Named("environmentstore"),
+	}
+}
+
+func (s *Store) Get(id string) (*model.Environment, bool) {
+	env, err := s.cache.Get(id)
+	if err == nil {
+		return env.(*model.Environment), true
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), s.apiTimeout)
+	defer cancel()
+
+	resp, err := s.apiClient.GetEnvironment(ctx, &pipedservice.GetEnvironmentRequest{
+		Id: id,
+	})
+	if err != nil {
+		s.logger.Warn("unable to get environment from control plane",
+			zap.String("env", id),
+			zap.Error(err),
+		)
+		return nil, false
+	}
+
+	if err := s.cache.Put(id, resp.Environment); err != nil {
+		s.logger.Warn("unable to put environment to cache", zap.Error(err))
+	}
+	return resp.Environment, true
+}

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

@@ -0,0 +1,15 @@
+// Copyright 2020 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 environmentstore

pkg/app/piped/cmd/piped/BUILD.bazel

@@ -12,6 +12,7 @@ go_library(
         "//pkg/app/piped/apistore/applicationstore:go_default_library",
         "//pkg/app/piped/apistore/commandstore:go_default_library",
         "//pkg/app/piped/apistore/deploymentstore:go_default_library",
+        "//pkg/app/piped/apistore/environmentstore:go_default_library",
         "//pkg/app/piped/chartrepo:go_default_library",
         "//pkg/app/piped/controller:go_default_library",
         "//pkg/app/piped/driftdetector:go_default_library",

pkg/app/piped/cmd/piped/piped.go

@@ -33,6 +33,7 @@ import (
 	"github.com/pipe-cd/pipe/pkg/app/piped/apistore/applicationstore"
 	"github.com/pipe-cd/pipe/pkg/app/piped/apistore/commandstore"
 	"github.com/pipe-cd/pipe/pkg/app/piped/apistore/deploymentstore"
+	"github.com/pipe-cd/pipe/pkg/app/piped/apistore/environmentstore"
 	"github.com/pipe-cd/pipe/pkg/app/piped/chartrepo"
 	"github.com/pipe-cd/pipe/pkg/app/piped/controller"
 	"github.com/pipe-cd/pipe/pkg/app/piped/driftdetector"
@@ -115,27 +116,9 @@ func (p *piped) run(ctx context.Context, t cli.Telemetry) (runErr error) {
 		t.Logger.Error("failed to initialize notifier", zap.Error(err))
 		return err
 	}
-	{
-		group.Go(func() error {
-			return notifier.Run(ctx)
-		})
-		notifier.Notify(model.Event{
-			Type: model.EventType_EVENT_PIPED_STARTED,
-			Metadata: &model.EventPipedStarted{
-				Id:      cfg.PipedID,
-				Version: version.Get().Version,
-			},
-		})
-		defer func() {
-			notifier.Notify(model.Event{
-				Type: model.EventType_EVENT_PIPED_STOPPED,
-				Metadata: &model.EventPipedStopped{
-					Id:      cfg.PipedID,
-					Version: version.Get().Version,
-				},
-			})
-		}()
-	}
+	group.Go(func() error {
+		return notifier.Run(ctx)
+	})
 
 	// Configure SSH config if needed.
 	if cfg.Git.ShouldConfigureSSHConfig() {
@@ -223,6 +206,13 @@ func (p *piped) run(ctx context.Context, t cli.Telemetry) (runErr error) {
 		}
 	}()
 
+	// Initialize environment store.
+	environmentStore := environmentstore.NewStore(
+		apiClient,
+		memorycache.NewTTLCache(ctx, 10*time.Minute, time.Minute),
+		t.Logger,
+	)
+
 	// Start running application store.
 	var applicationLister applicationstore.Lister
 	{
@@ -290,6 +280,7 @@ func (p *piped) run(ctx context.Context, t cli.Telemetry) (runErr error) {
 			deploymentLister,
 			commandLister,
 			applicationLister,
+			environmentStore,
 			livestatestore.LiveResourceLister{Getter: liveStateGetter},
 			notifier,
 			cfg,
@@ -310,6 +301,7 @@ func (p *piped) run(ctx context.Context, t cli.Telemetry) (runErr error) {
 			gitClient,
 			applicationLister,
 			commandLister,
+			environmentStore,
 			notifier,
 			cfg,
 			p.gracePeriod,

pkg/app/piped/controller/controller.go

@@ -76,6 +76,10 @@ type applicationLister interface {
 	Get(id string) (*model.Application, bool)
 }
 
+type environmentLister interface {
+	Get(id string) (*model.Environment, bool)
+}
+
 type liveResourceLister interface {
 	ListKubernetesAppLiveResources(cloudProvider, appID string) ([]provider.Manifest, bool)
 }
@@ -99,6 +103,7 @@ type controller struct {
 	deploymentLister   deploymentLister
 	commandLister      commandLister
 	applicationLister  applicationLister
+	environmentLister  environmentLister
 	liveResourceLister liveResourceLister
 	notifier           notifier
 	pipedConfig        *config.PipedSpec
@@ -137,6 +142,7 @@ func NewController(
 	deploymentLister deploymentLister,
 	commandLister commandLister,
 	applicationLister applicationLister,
+	environmentLister environmentLister,
 	liveResourceLister liveResourceLister,
 	notifier notifier,
 	pipedConfig *config.PipedSpec,
@@ -155,6 +161,7 @@ func NewController(
 		deploymentLister:   deploymentLister,
 		commandLister:      commandLister,
 		applicationLister:  applicationLister,
+		environmentLister:  environmentLister,
 		liveResourceLister: liveResourceLister,
 		notifier:           notifier,
 		appManifestsCache:  appManifestsCache,
@@ -347,8 +354,14 @@ func (c *controller) startNewPlanner(ctx context.Context, d *model.Deployment) (
 		}
 	}
 
+	var envName string
+	if env, ok := c.environmentLister.Get(d.EnvId); ok {
+		envName = env.Name
+	}
+
 	planner := newPlanner(
 		d,
+		envName,
 		commit,
 		workingDir,
 		c.apiClient,
@@ -479,9 +492,15 @@ func (c *controller) startNewScheduler(ctx context.Context, d *model.Deployment)
 	}
 	logger.Info("created working directory for scheduler", zap.String("working-dir", workingDir))
 
+	var envName string
+	if env, ok := c.environmentLister.Get(d.EnvId); ok {
+		envName = env.Name
+	}
+
 	// Create a new scheduler and append to the list for tracking.
 	scheduler := newScheduler(
 		d,
+		envName,
 		workingDir,
 		c.apiClient,
 		c.gitClient,

pkg/app/piped/controller/planner.go

@@ -42,6 +42,7 @@ import (
 type planner struct {
 	// Readonly deployment model.
 	deployment               *model.Deployment
+	envName                  string
 	lastSuccessfulCommitHash string
 	workingDir               string
 	apiClient                apiClient
@@ -63,6 +64,7 @@ type planner struct {
 
 func newPlanner(
 	d *model.Deployment,
+	envName string,
 	lastSuccessfulCommitHash string,
 	workingDir string,
 	apiClient apiClient,
@@ -84,6 +86,7 @@ func newPlanner(
 
 	p := &planner{
 		deployment:               d,
+		envName:                  envName,
 		lastSuccessfulCommitHash: lastSuccessfulCommitHash,
 		workingDir:               workingDir,
 		apiClient:                apiClient,
@@ -206,6 +209,7 @@ func (p *planner) reportDeploymentPlanned(ctx context.Context, runningCommitHash
 			Type: model.EventType_EVENT_DEPLOYMENT_PLANNED,
 			Metadata: &model.EventDeploymentPlanned{
 				Deployment: p.deployment,
+				EnvName:    p.envName,
 				Summary:    out.Summary,
 			},
 		})
@@ -243,6 +247,7 @@ func (p *planner) reportDeploymentFailed(ctx context.Context, reason string) err
 			Type: model.EventType_EVENT_DEPLOYMENT_FAILED,
 			Metadata: &model.EventDeploymentFailed{
 				Deployment: p.deployment,
+				EnvName:    p.envName,
 				Reason:     reason,
 			},
 		})
@@ -280,6 +285,7 @@ func (p *planner) reportDeploymentCancelled(ctx context.Context, commander, reas
 			Type: model.EventType_EVENT_DEPLOYMENT_CANCELLED,
 			Metadata: &model.EventDeploymentCancelled{
 				Deployment: p.deployment,
+				EnvName:    p.envName,
 				Commander:  commander,
 			},
 		})

pkg/app/piped/controller/scheduler.go

@@ -50,6 +50,7 @@ type repoStore interface {
 type scheduler struct {
 	// Readonly deployment model.
 	deployment         *model.Deployment
+	envName            string
 	workingDir         string
 	executorRegistry   registry.Registry
 	apiClient          apiClient
@@ -85,6 +86,7 @@ type scheduler struct {
 
 func newScheduler(
 	d *model.Deployment,
+	envName string,
 	workingDir string,
 	apiClient apiClient,
 	gitClient gitClient,
@@ -109,6 +111,7 @@ func newScheduler(
 
 	s := &scheduler{
 		deployment:           d,
+		envName:              envName,
 		workingDir:           workingDir,
 		executorRegistry:     registry.DefaultRegistry(),
 		apiClient:            apiClient,
@@ -594,6 +597,7 @@ func (s *scheduler) reportDeploymentCompleted(ctx context.Context, status model.
 				Type: model.EventType_EVENT_DEPLOYMENT_SUCCEEDED,
 				Metadata: &model.EventDeploymentSucceeded{
 					Deployment: s.deployment,
+					EnvName:    s.envName,
 				},
 			})
 
@@ -602,6 +606,7 @@ func (s *scheduler) reportDeploymentCompleted(ctx context.Context, status model.
 				Type: model.EventType_EVENT_DEPLOYMENT_FAILED,
 				Metadata: &model.EventDeploymentFailed{
 					Deployment: s.deployment,
+					EnvName:    s.envName,
 					Reason:     desc,
 				},
 			})
@@ -611,6 +616,7 @@ func (s *scheduler) reportDeploymentCompleted(ctx context.Context, status model.
 				Type: model.EventType_EVENT_DEPLOYMENT_CANCELLED,
 				Metadata: &model.EventDeploymentCancelled{
 					Deployment: s.deployment,
+					EnvName:    s.envName,
 					Commander:  cancelCommander,
 				},
 			})

pkg/app/piped/notifier/BUILD.bazel

@@ -13,7 +13,9 @@ go_library(
     deps = [
         "//pkg/config:go_default_library",
         "//pkg/model:go_default_library",
+        "//pkg/version:go_default_library",
         "@org_golang_x_sync//errgroup:go_default_library",
+        "@org_uber_go_atomic//:go_default_library",
         "@org_uber_go_zap//:go_default_library",
     ],
 )