Add apistore for caching the latest events

pkg/app/api/grpcapi/piped_api.go

@@ -753,6 +753,22 @@ func (a *PipedAPI) ListEvents(ctx context.Context, req *pipedservice.ListEventsR
 			},
 		},
 	}
+	switch req.Order {
+	case pipedservice.ListEventsOrder_FROM_OLDEST:
+		opts.Orders = []datastore.Order{
+			{
+				Field:     "CreatedAt",
+				Direction: datastore.Asc,
+			},
+		}
+	case pipedservice.ListEventsOrder_FROM_NEWEST:
+		opts.Orders = []datastore.Order{
+			{
+				Field:     "CreatedAt",
+				Direction: datastore.Desc,
+			},
+		}
+	}
 
 	events, err := a.eventStore.ListEvents(ctx, opts)
 	if err != nil {

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

@@ -362,9 +362,16 @@ message GetLatestEventResponse {
     pipe.model.Event event = 1 [(validate.rules).message.required = true];
 }
 
+enum ListEventsOrder {
+    NONE = 0;
+    FROM_OLDEST = 1;
+    FROM_NEWEST = 2;
+}
+
 message ListEventsRequest {
     int64 from = 1 [(validate.rules).int64.gt = 0];
     int64 to = 2 [(validate.rules).int64.gt = 0];
+    ListEventsOrder order = 3 [(validate.rules).enum.defined_only = true];
 }
 
 message ListEventsResponse {

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

@@ -0,0 +1,21 @@
+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/eventstore",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/app/api/service/pipedservice: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/eventstore/store.go

@@ -0,0 +1,154 @@
+// Copyright 2021 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 eventstore
+
+import (
+	"context"
+	"sync/atomic"
+	"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/model"
+)
+
+// Getter helps get an event. All objects returned here must be treated as read-only.
+type Getter interface {
+	// GetLatest returns the latest event that meets the given conditions.
+	GetLatest(ctx context.Context, name string, labels map[string]string) (*model.Event, bool)
+}
+
+type Store interface {
+	// Run starts syncing the event list with the control-plane.
+	Run(ctx context.Context) error
+	// Getter returns a getter for retrieving an event.
+	Getter() Getter
+}
+
+type apiClient interface {
+	GetLatestEvent(ctx context.Context, req *pipedservice.GetLatestEventRequest, opts ...grpc.CallOption) (*pipedservice.GetLatestEventResponse, error)
+	ListEvents(ctx context.Context, req *pipedservice.ListEventsRequest, opts ...grpc.CallOption) (*pipedservice.ListEventsResponse, error)
+}
+
+type store struct {
+	apiClient    apiClient
+	syncInterval time.Duration
+	gracePeriod  time.Duration
+	logger       *zap.Logger
+
+	// Mark that it has handled all events that was created before this value.
+	milestone int64
+	// The key is supposed to be event-definition ID, a string consists of name and labels.
+	// And the value is the address to the latest Event.
+	latestEventMap atomic.Value
+}
+
+const (
+	defaultSyncInterval = time.Minute
+)
+
+// NewStore creates a new event store instance.
+// This syncs with the control plane to keep the list of events for this runner up-to-date.
+func NewStore(apiClient apiClient, gracePeriod time.Duration, logger *zap.Logger) Store {
+	s := &store{
+		apiClient:    apiClient,
+		syncInterval: defaultSyncInterval,
+		gracePeriod:  gracePeriod,
+		logger:       logger.Named("event-store"),
+	}
+	s.latestEventMap.Store(make(map[string]*model.Event))
+	return s
+}
+
+// Run starts runner that periodically makes the Events in the cache up-to-date
+// by fetching from the control-plane.
+func (s *store) Run(ctx context.Context) error {
+	s.logger.Info("start running event store")
+
+	syncTicker := time.NewTicker(s.syncInterval)
+	defer syncTicker.Stop()
+
+	// Do first sync without waiting the first ticker.
+	s.milestone = time.Now().Add(-time.Hour).Unix()
+	s.sync(ctx)
+
+	for {
+		select {
+		case <-syncTicker.C:
+			s.sync(ctx)
+
+		case <-ctx.Done():
+			s.logger.Info("event store has been stopped")
+			return nil
+		}
+	}
+}
+
+// sync fetches a list of events inside the range between own milestone and the current local time.
+// Only this function takes responsibility for updating the cache.
+func (s *store) sync(ctx context.Context) error {
+	// TODO: Use UTC and let control-plane convert it into the control-plane's Local time
+	// Unexpected behavior can be happened if Piped uses different timezone from the control-plane.
+	to := time.Now().Unix()
+	resp, err := s.apiClient.ListEvents(ctx, &pipedservice.ListEventsRequest{
+		From:  s.milestone,
+		To:    to,
+		Order: pipedservice.ListEventsOrder_FROM_OLDEST,
+	})
+	if err != nil {
+		s.logger.Error("failed to list events", zap.Error(err))
+		return err
+	}
+	// Set this time's "to" as the next time's "from".
+	s.milestone = to
+
+	// Make the cache up-to-date by traversing events sorted by oldest first.
+	eventMap := s.latestEventMap.Load().(map[string]*model.Event)
+	for _, event := range resp.Events {
+		id := model.EventDefinitionID(event.Name, event.Labels)
+		eventMap[id] = event
+	}
+	s.latestEventMap.Store(eventMap)
+
+	return nil
+}
+
+func (s *store) Getter() Getter {
+	return s
+}
+
+func (s *store) GetLatest(ctx context.Context, name string, labels map[string]string) (*model.Event, bool) {
+	eventMap := s.latestEventMap.Load().(map[string]*model.Event)
+	id := model.EventDefinitionID(name, labels)
+	event, ok := eventMap[id]
+	if ok {
+		return event, true
+	}
+
+	// If not found in the cache, fetch from the control-plane.
+	resp, err := s.apiClient.GetLatestEvent(ctx, &pipedservice.GetLatestEventRequest{
+		Name:   name,
+		Labels: labels,
+	})
+	if err != nil {
+		s.logger.Error("failed to get the latest event", zap.Error(err))
+		return nil, false
+	}
+
+	// NOTE: Don't update the cache to prevent it from being overwritten by slightly older ones.
+	return resp.Event, true
+}

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

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

pkg/model/BUILD.bazel

@@ -53,6 +53,7 @@ go_library(
         "deployment.go",
         "docs.go",
         "environment.go",
+        "event.go",
         "filestore.go",
         "image_name.go",
         "imageprovider.go",
@@ -79,6 +80,7 @@ go_test(
     srcs = [
         "apikey_test.go",
         "common_test.go",
+        "event_test.go",
         "image_name_test.go",
         "model_test.go",
         "piped_test.go",

pkg/model/event.go

@@ -0,0 +1,43 @@
+// Copyright 2021 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 model
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+)
+
+// EventDefinitionID builds a unique identifier based on the given name and labels.
+// It returns the exact same string as long as both are the same.
+func EventDefinitionID(name string, labels map[string]string) string {
+	if len(labels) == 0 {
+		return name
+	}
+
+	var b strings.Builder
+	b.WriteString(name)
+
+	// Guarantee uniqueness by sorting by keys.
+	keys := make([]string, 0, len(labels))
+	for key := range labels {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	for _, key := range keys {
+		b.WriteString(fmt.Sprintf("/%s:%s", key, labels[key]))
+	}
+	return b.String()
+}

pkg/model/event_test.go

@@ -0,0 +1,63 @@
+// Copyright 2021 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 model
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestEventDefinitionID(t *testing.T) {
+	testcases := []struct {
+		testname string
+		name     string
+		labels   map[string]string
+		want     string
+	}{
+		{
+			testname: "no name and labels given",
+			want:     "",
+		},
+		{
+			testname: "no labels given",
+			name:     "name1",
+			want:     "name1",
+		},
+		{
+			testname: "no name given",
+			labels: map[string]string{
+				"key1": "value1",
+			},
+			want: "/key1:value1",
+		},
+		{
+			testname: "labels given",
+			name:     "name1",
+			labels: map[string]string{
+				"key1": "value",
+				"key2": "value",
+				"key3": "value",
+			},
+			want: "name1/key1:value/key2:value/key3:value",
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.testname, func(t *testing.T) {
+			got := EventDefinitionID(tc.name, tc.labels)
+			assert.Equal(t, tc.want, got)
+		})
+	}
+}