xds-server updates (googleforgames#437)

* xds-server updates

- Adds health probes and configurable logging
- Fixes a bug where we pass each endpoint as its own locality
  rather than all in the same locality
- Add support for routing tokens - if a routing token annotation
  is set, then the server configures the proxy with a
  `CaptureBytes=>TokenRouter` in the filter chain.
- Adds a simple dockerfile that can be used to build the server

* updates

* update docker,golangci-lint

* add /live

xds/README.md

@@ -90,6 +90,13 @@ The project has two binaries depending on the external source of configuration:
       The following annotations are currently supported:
       - **quilkin.dev/debug-packets**: If set to the value `true`, then a `Debug` filter will be
         added to the filter chain, causing all packets will be logged.
+      - **quilkin.dev/routing-token-suffix-size**: Sets the size (in number of bytes) of routing tokens appended to
+        packets. Extracted tokens will matched against available endpoints in order to figure out
+        where to send the associated packet.
+        Note that the token is stripped off the packet. This annotation cannot be provided together with
+        `quilkin.dev/routing-token-prefix-size`.
+      - **quilkin.dev/routing-token-prefix-size**: Works exactly the same as `quilkin.dev/routing-token-prefix-size`
+        with the difference that the token is a prefix on the packet rather than a suffix.
 
    As an example, the following runs the server against a cluster (using default kubeconfig configuration) where Quilkin pods run in the `quilkin` namespace and game-server pods run in the `gameservers` namespace:
 
@@ -102,6 +109,14 @@ The project has two binaries depending on the external source of configuration:
 
    > Note that currently, the server can only discover resources within a single cluster.
 
+   ##### Admin server
+
+   In addition the gRPC server, a http server (configurable via `--admin-port`is also started to serve administrative functionality.
+   The following endpoints are provided:
+   - `/ready`: Readiness probe that returns a 5xx if communication with the Kubernetes api is problematic.
+   - `/live`: Liveness probe that always returns a 200 response.
+   - `/metrics`: Exposes Prometheus metrics.
+
 
 [XDS]: https://www.envoyproxy.io/docs/envoy/latest/api-docs/xds_protocol
 [Kubernetes]: https://kubernetes.io/

xds/ci/.golangci.yaml

@@ -2,4 +2,18 @@ linters:
    enable:
    - gofmt
    - goimports
-   - revive
+   - revive
+   - bodyclose
+   - deadcode
+   - dupl
+   - exportloopref
+   - gocritic
+   - gocyclo
+   - govet
+   - megacheck
+   - misspell
+   - nakedret
+   - staticcheck
+   - structcheck
+   - unconvert
+   - unparam

xds/ci/Dockerfile

@@ -0,0 +1,34 @@
+#
+# Copyright 2021 Google LLC
+#
+# 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.
+
+FROM golang:1.16-alpine
+
+WORKDIR /app
+
+COPY go.mod ./
+COPY go.sum ./
+RUN go mod download
+
+COPY cmd/ cmd/
+COPY pkg/ pkg/
+
+RUN go build -o /server cmd/controller.go
+
+FROM gcr.io/distroless/static:nonroot
+
+WORKDIR /app
+COPY --from=0 /server ./
+
+ENTRYPOINT [ "./server" ]

xds/cmd/controller.go

@@ -18,8 +18,10 @@ package main
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"os/signal"
+	"strings"
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -31,6 +33,7 @@ import (
 
 	agones "agones.dev/agones/pkg/client/clientset/versioned"
 
+	k8sv1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/clock"
 
 	"agones.dev/agones/pkg/client/informers/externalversions"
@@ -51,14 +54,35 @@ type flags struct {
 	GameServersNamespace    string        `name:"game-server-namespace" help:"Namespace under which the game-servers run." default:"gameservers"`
 	GameServersPollInterval time.Duration `name:"game-server-poll-interval" help:"How long to wait in-between checking for game-server updates." default:"1s"`
 	ProxyPollInterval       time.Duration `name:"proxy-interval" help:"How long to wait in-between checking for proxy updates." default:"1s"`
+	AdminPort               int16         `name:"admin-port" help:"Admin server listening port." default:"18090"`
+	LogLevel                string        `name:"log-level" help:"Log level, one of trace, debug, info, warn error, fatal" default:"info"`
+}
+
+func getLogLevel(value string) (log.Level, error) {
+	switch strings.ToLower(value) {
+	case "trace":
+		return log.TraceLevel, nil
+	case "debug":
+		return log.DebugLevel, nil
+	case "info":
+		return log.InfoLevel, nil
+	case "warn":
+		return log.WarnLevel, nil
+	case "error":
+		return log.ErrorLevel, nil
+	case "fatal":
+		return log.FatalLevel, nil
+	default:
+		return 0, fmt.Errorf("invalid log level '%s'", value)
+	}
 }
 
 func createAgonesClusterProvider(
 	ctx context.Context,
 	logger *log.Logger,
 	k8sConfig *rest.Config,
 	flags *flags,
-) cluster.Provider {
+) (cluster.Provider, server.HealthCheck) {
 	agonesClient, err := agones.NewForConfig(k8sConfig)
 	if err != nil {
 		log.WithError(err).Fatal("failed to create Agones clientset")
@@ -76,15 +100,38 @@ func createAgonesClusterProvider(
 	return agonescluster.NewProvider(logger, gameServerLister, agonescluster.Config{
 		GameServersNamespace:    flags.GameServersNamespace,
 		GameServersPollInterval: flags.GameServersPollInterval,
-	})
+	}), createAgonesProviderHealthCheck(flags.GameServersNamespace, agonesClient)
+}
+
+func createAgonesProviderHealthCheck(
+	gameServersNamespace string,
+	agonesClient *agones.Clientset,
+) server.HealthCheck {
+	return func(ctx context.Context) error {
+		if _, err := agonesClient.
+			AgonesV1().
+			RESTClient().
+			Get().
+			AbsPath("/healthz").
+			DoRaw(ctx); err != nil {
+			return fmt.Errorf("agones cluster provider failed to reach k8s api: %w", err)
+		}
+		if _, err := agonesClient.
+			AgonesV1().
+			GameServers(gameServersNamespace).
+			List(ctx, k8sv1.ListOptions{Limit: 1}); err != nil {
+			return fmt.Errorf("agones cluster provider failed to list GameServers objects: %w", err)
+		}
+		return nil
+	}
 }
 
 func createFilterChainProvider(
 	ctx context.Context,
 	logger *log.Logger,
 	k8sClient *kubernetes.Clientset,
 	flags *flags,
-) filterchain.Provider {
+) (filterchain.Provider, server.HealthCheck) {
 
 	informerFactory := informers.NewSharedInformerFactoryWithOptions(
 		k8sClient,
@@ -103,16 +150,36 @@ func createFilterChainProvider(
 		logger,
 		clock.RealClock{},
 		podLister,
-		flags.ProxyPollInterval)
+		flags.ProxyPollInterval), createFilterChainProviderHealthCheck(flags.ProxyNamespace, k8sClient)
+}
+
+func createFilterChainProviderHealthCheck(
+	podNamespace string,
+	k8sClient *kubernetes.Clientset,
+) server.HealthCheck {
+	return func(ctx context.Context) error {
+		if _, err := k8sClient.
+			CoreV1().
+			Pods(podNamespace).
+			List(ctx, metav1.ListOptions{Limit: 1}); err != nil {
+			return fmt.Errorf("k8s filter chain provider failed to list Pod objects: %w", err)
+		}
+		return nil
+	}
 }
 
 func main() {
 	var flags flags
 	kong.Parse(&flags)
 
+	logLevel, err := getLogLevel(flags.LogLevel)
+	if err != nil {
+		log.Fatal(err)
+	}
+
 	logger := &log.Logger{}
 	logger.SetOutput(os.Stdout)
-	logger.SetLevel(log.DebugLevel)
+	logger.SetLevel(logLevel)
 	logger.SetFormatter(&log.JSONFormatter{})
 
 	ctx, shutdown := context.WithCancel(context.Background())
@@ -127,13 +194,13 @@ func main() {
 	if err != nil {
 		log.WithError(err).Fatal("failed to create k8s client")
 	}
-	clusterProvider := createAgonesClusterProvider(ctx, logger, k8sConfig, &flags)
+	clusterProvider, clusterHealthCheck := createAgonesClusterProvider(ctx, logger, k8sConfig, &flags)
 	clusterCh, err := clusterProvider.Run(ctx)
 	if err != nil {
 		log.WithError(err).Fatal("failed to create start cluster provider")
 	}
 
-	filterChainProvider := createFilterChainProvider(ctx, logger, k8sClient, &flags)
+	filterChainProvider, filterChainHealthCheck := createFilterChainProvider(ctx, logger, k8sClient, &flags)
 	filterChainCh := filterChainProvider.Run(ctx)
 
 	snapshotUpdater := snapshot.NewUpdater(
@@ -145,7 +212,16 @@ func main() {
 	snapshotCache := snapshotUpdater.GetSnapshotCache()
 	go snapshotUpdater.Run(ctx)
 
-	srv := server.New(logger, flags.Port, snapshotCache, nil)
+	srv := server.New(
+		logger,
+		flags.Port,
+		snapshotCache,
+		nil,
+		[]server.HealthCheck{
+			clusterHealthCheck,
+			filterChainHealthCheck,
+		},
+		flags.AdminPort)
 	if err := srv.Run(ctx); err != nil {
 		logger.WithError(err).Fatal("failed to start server")
 	}

xds/cmd/file/file.go

@@ -64,7 +64,7 @@ func main() {
 	snapshotCache := snapshotUpdater.GetSnapshotCache()
 	go snapshotUpdater.Run(ctx)
 
-	srv := server.New(logger, flags.Port, snapshotCache, proxyIDCh)
+	srv := server.New(logger, flags.Port, snapshotCache, proxyIDCh, []server.HealthCheck{}, 8008)
 	if err := srv.Run(ctx); err != nil {
 		logger.WithError(err).Fatal("failed to start server")
 	}
@@ -87,5 +87,4 @@ func main() {
 	case <-ctx.Done():
 		logger.Info("Shutdown.")
 	}
-
 }

xds/go.mod

@@ -10,14 +10,14 @@ require (
 	github.com/fsnotify/fsnotify v1.4.9
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang/protobuf v1.5.2
+	github.com/prometheus/client_golang v1.7.1
 	github.com/sirupsen/logrus v1.8.1
 	github.com/stretchr/testify v1.7.0
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 	google.golang.org/grpc v1.36.0
 	google.golang.org/protobuf v1.26.0
-	gopkg.in/yaml.v2 v2.4.0 // indirect
 	k8s.io/api v0.22.3
 	k8s.io/apimachinery v0.22.3
 	k8s.io/client-go v0.22.3
-	k8s.io/sample-controller v0.22.3
 	sigs.k8s.io/yaml v1.2.0
 )