messenger: Add handler decorator that is resource aware (#2555)

If a resource is not healthy the handler should not be executed, instead an error should be returned.
This can be used to make handlers resource aware so that they don't execute when e.g. a database is not available.

go/lib/infra/BUILD.bazel

@@ -1,4 +1,4 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
@@ -18,9 +18,23 @@ go_library(
         "//go/lib/ctrl/ifid:go_default_library",
         "//go/lib/ctrl/path_mgmt:go_default_library",
         "//go/lib/ctrl/seg:go_default_library",
+        "//go/lib/log:go_default_library",
         "//go/lib/prom:go_default_library",
         "//go/lib/scrypto/cert:go_default_library",
         "//go/lib/scrypto/trc:go_default_library",
         "//go/proto:go_default_library",
     ],
 )
+
+go_test(
+    name = "go_default_test",
+    srcs = ["common_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//go/lib/ctrl/ack:go_default_library",
+        "//go/lib/infra/mock_infra:go_default_library",
+        "//go/proto:go_default_library",
+        "@com_github_golang_mock//gomock:go_default_library",
+        "@com_github_smartystreets_goconvey//convey:go_default_library",
+    ],
+)

go/lib/infra/common.go

@@ -27,6 +27,7 @@ import (
 	"github.com/scionproto/scion/go/lib/ctrl/ifid"
 	"github.com/scionproto/scion/go/lib/ctrl/path_mgmt"
 	"github.com/scionproto/scion/go/lib/ctrl/seg"
+	"github.com/scionproto/scion/go/lib/log"
 	"github.com/scionproto/scion/go/lib/scrypto/cert"
 	"github.com/scionproto/scion/go/lib/scrypto/trc"
 	"github.com/scionproto/scion/go/proto"
@@ -246,6 +247,43 @@ func (mt MessageType) MetricLabel() string {
 	}
 }
 
+// ResourceHealth indicates the health of a resource. A resource could for example be a database.
+// The resource health can be added to a handler, so that the handler only replies if all it's
+// resources are healthy.
+type ResourceHealth interface {
+	// Name returns the name of this resource.
+	Name() string
+	// IsHealthy returns whether the resource is considered healthy currently.
+	// This method must not be blocking and should have the result cached and return ~immediately.
+	IsHealthy() bool
+}
+
+// NewResourceAwareHandler creates a decorated handler that calls the underlying handler if all
+// resources are healthy, otherwise it replies with an error message.
+func NewResourceAwareHandler(handler Handler, resources ...ResourceHealth) Handler {
+	return HandlerFunc(func(r *Request) *HandlerResult {
+		ctx := r.Context()
+		for _, resource := range resources {
+			if !resource.IsHealthy() {
+				logger := log.FromCtx(ctx)
+				rwriter, ok := ResponseWriterFromContext(ctx)
+				if !ok {
+					logger.Error("No response writer found")
+					return MetricsErrInternal
+				}
+				logger.Warn("Resource not healthy, can't handle request",
+					"resource", resource.Name())
+				rwriter.SendAckReply(ctx, &ack.Ack{
+					Err:     proto.Ack_ErrCode_reject,
+					ErrDesc: fmt.Sprintf("Resource %s not healthy", resource.Name()),
+				})
+				return MetricsErrInternal
+			}
+		}
+		return handler.Handle(r)
+	})
+}
+
 type Messenger interface {
 	SendAck(ctx context.Context, msg *ack.Ack, a net.Addr, id uint64) error
 	// GetTRC sends a cert_mgmt.TRCReq request to address a, blocks until it receives a

go/lib/infra/common_test.go

@@ -0,0 +1,63 @@
+// Copyright 2019 Anapaya Systems
+//
+// 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 infra_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	. "github.com/smartystreets/goconvey/convey"
+
+	"github.com/scionproto/scion/go/lib/ctrl/ack"
+	"github.com/scionproto/scion/go/lib/infra"
+	"github.com/scionproto/scion/go/lib/infra/mock_infra"
+	"github.com/scionproto/scion/go/proto"
+)
+
+var _ infra.ResourceHealth = (*mockResource)(nil)
+
+type mockResource struct {
+	name    string
+	healthy bool
+}
+
+func (r *mockResource) Name() string {
+	return r.name
+}
+
+func (r *mockResource) IsHealthy() bool {
+	return r.healthy
+}
+
+func TestResourceHealth(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	Convey("Unhealthy resource results in error replied", t, func() {
+		handler := infra.HandlerFunc(func(r *infra.Request) *infra.HandlerResult {
+			return nil
+		})
+		rHandler := infra.NewResourceAwareHandler(handler,
+			&mockResource{name: "tstFail", healthy: false})
+		rwMock := mock_infra.NewMockResponseWriter(ctrl)
+		ctx := infra.NewContextWithResponseWriter(context.Background(), rwMock)
+		rwMock.EXPECT().SendAckReply(gomock.Eq(ctx), gomock.Eq(&ack.Ack{
+			Err:     proto.Ack_ErrCode_reject,
+			ErrDesc: "Resource tstFail not healthy",
+		}))
+		req := infra.NewRequest(ctx, nil, nil, nil, 1)
+		rHandler.Handle(req)
+	})
+}