Start livestatereporter on pipedv1

pkg/app/pipedv1/apistore/applicationstore/store.go

@@ -32,10 +32,10 @@
 	// List lists all applications that should be handled by this piped.
 	// All disabled applications will be ignored.
 	List() []*model.Application
-	// ListByPlatformProvider lists all applications for a given cloud provider name.
-	ListByPlatformProvider(name string) []*model.Application
 	// Get retrieves a specifiec deployment for the given id.
 	Get(id string) (*model.Application, bool)
+	// ListByPluginName lists all applications for a given plugin name.
+	ListByPluginName(name string) []*model.Application
 }
 
 type apiClient interface {
@@ -127,25 +127,6 @@
 	return apps.([]*model.Application)
 }
 
-// ListByPlatformProvider lists all applications for a given platform provider name.
-func (s *store) ListByPlatformProvider(name string) []*model.Application {
-	list := s.applicationList.Load()
-	if list == nil {
-		return nil
-	}
-
-	var (
-		apps = list.([]*model.Application)
-		out  = make([]*model.Application, 0, len(apps))
-	)
-	for _, app := range apps {
-		if app.PlatformProvider == name {
-			out = append(out, app)
-		}
-	}
-	return out
-}
-
 // Get retrieves a specific deployment for the given id.
 func (s *store) Get(id string) (*model.Application, bool) {
 	apps := s.applicationMap.Load()
@@ -156,3 +137,25 @@
 	app, ok := apps.(map[string]*model.Application)[id]
 	return app, ok
 }
+
+// ListByPluginName lists all applications for a given plugin name.
+func (s *store) ListByPluginName(name string) []*model.Application {
+	apps := s.applicationList.Load()
+	if apps == nil {
+		return nil
+	}
+
+	out := make([]*model.Application, 0)
+	list := apps.([]*model.Application)
+
+	for _, app := range list {
+		for _, p := range app.Plugins {
+			if p == name {
+				out = append(out, app)
+				break
+			}
+		}
+	}
+
+	return out
+}

pkg/app/pipedv1/apistore/applicationstore/store_test.go

@@ -0,0 +1,89 @@
+// Copyright 2024 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 applicationstore
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/atomic"
+	"go.uber.org/zap/zaptest"
+
+	"github.com/pipe-cd/pipecd/pkg/model"
+)
+
+func TestListByPluginName(t *testing.T) {
+	logger := zaptest.NewLogger(t)
+
+	tests := []struct {
+		name       string
+		storedApps []*model.Application
+		plugin     string
+		expected   []*model.Application
+	}{
+		{
+			name:       "There is no stored application",
+			storedApps: []*model.Application{},
+			plugin:     "plugin-a",
+			expected:   []*model.Application{},
+		},
+		{
+			name: "no matching",
+			storedApps: []*model.Application{
+				{Id: "app-1", Plugins: []string{"plugin-b"}},
+				{Id: "app-2", Plugins: []string{"plugin-c"}},
+			},
+			plugin:   "plugin-a",
+			expected: []*model.Application{},
+		},
+		{
+			name: "one matched application",
+			storedApps: []*model.Application{
+				{Id: "app-1", Plugins: []string{"plugin-a", "plugin-b"}},
+				{Id: "app-2", Plugins: []string{"plugin-b"}},
+			},
+			plugin: "plugin-a",
+			expected: []*model.Application{
+				{Id: "app-1", Plugins: []string{"plugin-a", "plugin-b"}},
+			},
+		},
+		{
+			name: "matched some applications",
+			storedApps: []*model.Application{
+				{Id: "app-1", Plugins: []string{"plugin-a", "plugin-b"}},
+				{Id: "app-2", Plugins: []string{"plugin-a"}},
+				{Id: "app-3", Plugins: []string{"plugin-b"}},
+			},
+			plugin: "plugin-a",
+			expected: []*model.Application{
+				{Id: "app-1", Plugins: []string{"plugin-a", "plugin-b"}},
+				{Id: "app-2", Plugins: []string{"plugin-a"}},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &store{
+				applicationList: atomic.Value{},
+				logger:          logger,
+			}
+			s.applicationList.Store(tt.storedApps)
+
+			got := s.ListByPluginName(tt.plugin)
+			assert.Equal(t, tt.expected, got)
+		})
+	}
+}

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

@@ -60,6 +60,7 @@
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/controller"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/controller/controllermetrics"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/eventwatcher"
+	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/livestatereporter"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/notifier"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/statsreporter"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/trigger"
@@ -349,6 +350,8 @@
 
 	// Make grpc clients to connect to plugins.
 	pluginClis := make([]pluginapi.PluginClient, 0, len(cfg.Plugins))
+	nameBasedPluginClis := make(map[string]pluginapi.PluginClient, len(cfg.Plugins))
+
 	options := []rpcclient.DialOption{
 		rpcclient.WithBlock(),
 		rpcclient.WithInsecure(),
@@ -359,11 +362,15 @@
 			input.Logger.Error("failed to create client to connect plugin", zap.String("plugin", plg.Name), zap.Error(err))
 		}
 		pluginClis = append(pluginClis, cli)
+		nameBasedPluginClis[plg.Name] = cli
 	}
 
 	// Start running application live state reporter.
 	{
-		// TODO: Implement the live state reporter controller.
+		r := livestatereporter.NewReporter(applicationLister, apiClient, nameBasedPluginClis, input.Logger)
+		group.Go(func() error {
+			return r.Run(ctx)
+		})
 	}
 
 	// Start running application application drift detector.