K8s: add a remote services file config option to specify aggregation config (#83646)

hack/make-aggregator-pki.sh

@@ -7,6 +7,21 @@ set -o pipefail
 rm -rf data/grafana-aggregator
 
 mkdir -p data/grafana-aggregator
+
 openssl req -nodes -new -x509 -keyout data/grafana-aggregator/ca.key -out data/grafana-aggregator/ca.crt
-openssl req -out data/grafana-aggregator/client.csr -new -newkey rsa:4096 -nodes -keyout data/grafana-aggregator/client.key -subj "/CN=development/O=system:masters"
-openssl x509 -req -days 365 -in data/grafana-aggregator/client.csr -CA data/grafana-aggregator/ca.crt -CAkey data/grafana-aggregator/ca.key -set_serial 01 -sha256 -out data/grafana-aggregator/client.crt
+openssl req -out data/grafana-aggregator/client.csr -new -newkey rsa:4096 -nodes -keyout data/grafana-aggregator/client.key \
+  -subj "/CN=development/O=system:masters" \
+  -addext "extendedKeyUsage = clientAuth"
+openssl x509 -req -days 365 -in data/grafana-aggregator/client.csr -CA data/grafana-aggregator/ca.crt -CAkey data/grafana-aggregator/ca.key \
+  -set_serial 01 \
+  -sha256 -out data/grafana-aggregator/client.crt \
+  -copy_extensions=copyall
+
+openssl req -out data/grafana-aggregator/server.csr -new -newkey rsa:4096 -nodes -keyout data/grafana-aggregator/server.key \
+  -subj "/CN=localhost/O=aggregated" \
+  -addext "subjectAltName = DNS:v0alpha1.example.grafana.app.default.svc,DNS:localhost" \
+  -addext "extendedKeyUsage = serverAuth, clientAuth"
+openssl x509 -req -days 365 -in data/grafana-aggregator/server.csr -CA data/grafana-aggregator/ca.crt -CAkey data/grafana-aggregator/ca.key \
+  -set_serial 02 \
+  -sha256 -out data/grafana-aggregator/server.crt \
+  -copy_extensions=copyall

pkg/registry/apis/service/register.go

@@ -43,6 +43,13 @@ func (b *ServiceAPIBuilder) GetGroupVersion() schema.GroupVersion {
 	return service.SchemeGroupVersion
 }
 
+func addKnownTypes(scheme *runtime.Scheme, gv schema.GroupVersion) {
+	scheme.AddKnownTypes(gv,
+		&service.ExternalName{},
+		&service.ExternalNameList{},
+	)
+}
+
 func (b *ServiceAPIBuilder) InstallSchema(scheme *runtime.Scheme) error {
 	gv := service.SchemeGroupVersion
 	err := service.AddToScheme(scheme)
@@ -53,10 +60,10 @@ func (b *ServiceAPIBuilder) InstallSchema(scheme *runtime.Scheme) error {
 	// Link this version to the internal representation.
 	// This is used for server-side-apply (PATCH), and avoids the error:
 	//   "no kind is registered for the type"
-	// addKnownTypes(scheme, schema.GroupVersion{
-	// 	Group:   service.GROUP,
-	// 	Version: runtime.APIVersionInternal,
-	// })
+	addKnownTypes(scheme, schema.GroupVersion{
+		Group:   service.GROUP,
+		Version: runtime.APIVersionInternal,
+	})
 	metav1.AddToGroupVersion(scheme, gv)
 	return scheme.SetVersionPriority(gv)
 }

pkg/services/apiserver/aggregator/README.md

@@ -77,7 +77,7 @@ configuration overwrites on startup.
 4. In another tab, apply the manifests: 
   ```shell
   export KUBECONFIG=$PWD/data/grafana-apiserver/grafana.kubeconfig
-  kubectl apply -f ./pkg/services/apiserver/aggregator/examples/
+  kubectl apply -f ./pkg/services/apiserver/aggregator/examples/manual-test/
   # SAMPLE OUTPUT
   # apiservice.apiregistration.k8s.io/v0alpha1.example.grafana.app created
   # externalname.service.grafana.app/example-apiserver created
@@ -92,6 +92,8 @@ configuration overwrites on startup.
   go run ./pkg/cmd/grafana apiserver \
     --runtime-config=example.grafana.app/v0alpha1=true \
     --secure-port 7443 \
+    --tls-cert-file $PWD/data/grafana-aggregator/server.crt \
+    --tls-private-key-file $PWD/data/grafana-aggregator/server.key \ 
     --requestheader-client-ca-file=$PWD/data/grafana-aggregator/ca.crt \
     --requestheader-extra-headers-prefix=X-Remote-Extra- \
     --requestheader-group-headers=X-Remote-Group \
@@ -110,3 +112,16 @@ configuration overwrites on startup.
   ```shell
   kubectl delete -f ./pkg/services/apiserver/aggregator/examples/
   ```
+
+## Testing auto-registration of remote services locally
+
+A sample aggregation config for remote services is provided under [conf](../../../../conf/aggregation/apiservices.yaml). Provided, you have the following setup in your custom.ini, the apiserver will
+register your remotely running services on startup.
+
+```ini
+; in custom.ini
+; the bundle is only used when not in dev mode
+apiservice_ca_bundle_file = ./data/grafana-aggregator/ca.crt
+
+remote_services_file = ./pkg/services/apiserver/aggregator/examples/autoregister/apiservices.yaml
+```

pkg/services/apiserver/aggregator/aggregator.go

@@ -12,13 +12,18 @@
 package aggregator
 
 import (
+	"context"
 	"crypto/tls"
 	"fmt"
+	"io"
 	"net/http"
+	"os"
 	"strings"
 	"sync"
 	"time"
 
+	servicev0alpha1 "github.com/grafana/grafana/pkg/apis/service/v0alpha1"
+	"gopkg.in/yaml.v3"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilnet "k8s.io/apimachinery/pkg/util/net"
@@ -37,19 +42,68 @@ import (
 	apiregistrationInformers "k8s.io/kube-aggregator/pkg/client/informers/externalversions/apiregistration/v1"
 	"k8s.io/kube-aggregator/pkg/controllers/autoregister"
 
+	servicev0alpha1applyconfiguration "github.com/grafana/grafana/pkg/generated/applyconfiguration/service/v0alpha1"
 	serviceclientset "github.com/grafana/grafana/pkg/generated/clientset/versioned"
 	informersv0alpha1 "github.com/grafana/grafana/pkg/generated/informers/externalversions"
 	"github.com/grafana/grafana/pkg/services/apiserver/options"
 )
 
-func CreateAggregatorConfig(commandOptions *options.Options, sharedConfig genericapiserver.RecommendedConfig) (*aggregatorapiserver.Config, informersv0alpha1.SharedInformerFactory, error) {
+func readCABundlePEM(path string, devMode bool) ([]byte, error) {
+	if devMode {
+		return nil, nil
+	}
+
+	// We can ignore the gosec G304 warning on this one because `path` comes
+	// from Grafana configuration (commandOptions.AggregatorOptions.APIServiceCABundleFile)
+	//nolint:gosec
+	f, err := os.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	defer func() {
+		if err := f.Close(); err != nil {
+			klog.Errorf("error closing remote services file: %s", err)
+		}
+	}()
+
+	return io.ReadAll(f)
+}
+
+func readRemoteServices(path string) ([]RemoteService, error) {
+	// We can ignore the gosec G304 warning on this one because `path` comes
+	// from Grafana configuration (commandOptions.AggregatorOptions.RemoteServicesFile)
+	//nolint:gosec
+	f, err := os.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	defer func() {
+		if err := f.Close(); err != nil {
+			klog.Errorf("error closing remote services file: %s", err)
+		}
+	}()
+
+	rawRemoteServices, err := io.ReadAll(f)
+	if err != nil {
+		return nil, err
+	}
+
+	remoteServices := make([]RemoteService, 0)
+	if err := yaml.Unmarshal(rawRemoteServices, &remoteServices); err != nil {
+		return nil, err
+	}
+
+	return remoteServices, nil
+}
+
+func CreateAggregatorConfig(commandOptions *options.Options, sharedConfig genericapiserver.RecommendedConfig, externalNamesNamespace string) (*Config, error) {
 	// Create a fake clientset and informers for the k8s v1 API group.
 	// These are not used in grafana's aggregator because v1 APIs are not available.
 	fakev1Informers := informers.NewSharedInformerFactory(fake.NewSimpleClientset(), 10*time.Minute)
 
 	serviceClient, err := serviceclientset.NewForConfig(sharedConfig.LoopbackClientConfig)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 	sharedInformerFactory := informersv0alpha1.NewSharedInformerFactory(
 		serviceClient,
@@ -74,13 +128,35 @@ func CreateAggregatorConfig(commandOptions *options.Options, sharedConfig generi
 	}
 
 	if err := commandOptions.AggregatorOptions.ApplyTo(aggregatorConfig, commandOptions.RecommendedOptions.Etcd, commandOptions.StorageOptions.DataPath); err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
-	return aggregatorConfig, sharedInformerFactory, nil
+	// Exit early, if no remote services file is configured
+	if commandOptions.AggregatorOptions.RemoteServicesFile == "" {
+		return NewConfig(aggregatorConfig, sharedInformerFactory, nil), nil
+	}
+
+	caBundlePEM, err := readCABundlePEM(commandOptions.AggregatorOptions.APIServiceCABundleFile, commandOptions.ExtraOptions.DevMode)
+	if err != nil {
+		return nil, err
+	}
+	remoteServices, err := readRemoteServices(commandOptions.AggregatorOptions.RemoteServicesFile)
+	if err != nil {
+		return nil, err
+	}
+
+	remoteServicesConfig := &RemoteServicesConfig{
+		InsecureSkipTLSVerify:  commandOptions.ExtraOptions.DevMode,
+		ExternalNamesNamespace: externalNamesNamespace,
+		CABundle:               caBundlePEM,
+		Services:               remoteServices,
+		serviceClientSet:       serviceClient,
+	}
+
+	return NewConfig(aggregatorConfig, sharedInformerFactory, remoteServicesConfig), nil
 }
 
-func CreateAggregatorServer(aggregatorConfig *aggregatorapiserver.Config, sharedInformerFactory informersv0alpha1.SharedInformerFactory, delegateAPIServer genericapiserver.DelegationTarget) (*aggregatorapiserver.APIAggregator, error) {
+func CreateAggregatorServer(aggregatorConfig *aggregatorapiserver.Config, sharedInformerFactory informersv0alpha1.SharedInformerFactory, remoteServicesConfig *RemoteServicesConfig, delegateAPIServer genericapiserver.DelegationTarget) (*aggregatorapiserver.APIAggregator, error) {
 	completedConfig := aggregatorConfig.Complete()
 	aggregatorServer, err := completedConfig.NewWithDelegate(delegateAPIServer)
 	if err != nil {
@@ -111,6 +187,27 @@ func CreateAggregatorServer(aggregatorConfig *aggregatorapiserver.Config, shared
 		return nil, err
 	}
 
+	if remoteServicesConfig != nil {
+		addRemoteAPIServicesToRegister(remoteServicesConfig, autoRegistrationController)
+		externalNames := getRemoteExternalNamesToRegister(remoteServicesConfig)
+		err = aggregatorServer.GenericAPIServer.AddPostStartHook("grafana-apiserver-remote-autoregistration", func(_ genericapiserver.PostStartHookContext) error {
+			namespacedClient := remoteServicesConfig.serviceClientSet.ServiceV0alpha1().ExternalNames(remoteServicesConfig.ExternalNamesNamespace)
+			for _, externalName := range externalNames {
+				_, err := namespacedClient.Apply(context.Background(), externalName, metav1.ApplyOptions{
+					FieldManager: "grafana-aggregator",
+					Force:        true,
+				})
+				if err != nil {
+					return err
+				}
+			}
+			return nil
+		})
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	err = aggregatorServer.GenericAPIServer.AddBootSequenceHealthChecks(
 		makeAPIServiceAvailableHealthCheck(
 			"autoregister-completion",
@@ -240,6 +337,51 @@ var APIVersionPriorities = map[schema.GroupVersion]Priority{
 	// Version can be set to 9 (to have space around) for a new group.
 }
 
+func addRemoteAPIServicesToRegister(config *RemoteServicesConfig, registration autoregister.AutoAPIServiceRegistration) {
+	for i, service := range config.Services {
+		port := service.Port
+		apiService := &v1.APIService{
+			ObjectMeta: metav1.ObjectMeta{Name: service.Version + "." + service.Group},
+			Spec: v1.APIServiceSpec{
+				Group:                 service.Group,
+				Version:               service.Version,
+				InsecureSkipTLSVerify: config.InsecureSkipTLSVerify,
+				CABundle:              config.CABundle,
+				// TODO: Group priority minimum of 1000 more than for local services, figure out a better story
+				// when we have multiple versions, potentially running in heterogeneous ways (local and remote)
+				GroupPriorityMinimum: 16000,
+				VersionPriority:      1 + int32(i),
+				Service: &v1.ServiceReference{
+					Name:      service.Version + "." + service.Group,
+					Namespace: config.ExternalNamesNamespace,
+					Port:      &port,
+				},
+			},
+		}
+
+		registration.AddAPIServiceToSyncOnStart(apiService)
+	}
+}
+
+func getRemoteExternalNamesToRegister(config *RemoteServicesConfig) []*servicev0alpha1applyconfiguration.ExternalNameApplyConfiguration {
+	externalNames := make([]*servicev0alpha1applyconfiguration.ExternalNameApplyConfiguration, 0)
+
+	for _, service := range config.Services {
+		host := service.Host
+		name := service.Version + "." + service.Group
+		externalName := &servicev0alpha1applyconfiguration.ExternalNameApplyConfiguration{}
+		externalName.WithAPIVersion(servicev0alpha1.SchemeGroupVersion.String())
+		externalName.WithKind("ExternalName")
+		externalName.WithName(name)
+		externalName.WithSpec(&servicev0alpha1applyconfiguration.ExternalNameSpecApplyConfiguration{
+			Host: &host,
+		})
+		externalNames = append(externalNames, externalName)
+	}
+
+	return externalNames
+}
+
 func apiServicesToRegister(delegateAPIServer genericapiserver.DelegationTarget, registration autoregister.AutoAPIServiceRegistration) []*v1.APIService {
 	apiServices := []*v1.APIService{}
 

pkg/services/apiserver/aggregator/config.go

@@ -0,0 +1,37 @@
+package aggregator
+
+import (
+	serviceclientset "github.com/grafana/grafana/pkg/generated/clientset/versioned"
+	informersv0alpha1 "github.com/grafana/grafana/pkg/generated/informers/externalversions"
+	aggregatorapiserver "k8s.io/kube-aggregator/pkg/apiserver"
+)
+
+type RemoteService struct {
+	Group   string `yaml:"group"`
+	Version string `yaml:"version"`
+	Host    string `yaml:"host"`
+	Port    int32  `yaml:"port"`
+}
+
+type RemoteServicesConfig struct {
+	ExternalNamesNamespace string
+	InsecureSkipTLSVerify  bool
+	CABundle               []byte
+	Services               []RemoteService
+	serviceClientSet       *serviceclientset.Clientset
+}
+
+type Config struct {
+	KubeAggregatorConfig *aggregatorapiserver.Config
+	Informers            informersv0alpha1.SharedInformerFactory
+	RemoteServicesConfig *RemoteServicesConfig
+}
+
+// remoteServices may be nil, when not using aggregation
+func NewConfig(aggregator *aggregatorapiserver.Config, informers informersv0alpha1.SharedInformerFactory, remoteServices *RemoteServicesConfig) *Config {
+	return &Config{
+		aggregator,
+		informers,
+		remoteServices,
+	}
+}

pkg/services/apiserver/aggregator/examples/autoregister/apiservices.yaml

@@ -0,0 +1,14 @@
+# NOTE: dev-mode only and governed by presence of non-empty value for cfg["grafana-apiserver"]["remote_services_file"]
+# List of sample multi-tenant services to aggregate on startup
+- group: example.grafana.app
+  version: v0alpha1
+  host: localhost
+  port: 7443
+- group: query.grafana.app
+  version: v0alpha1
+  host: localhost
+  port: 7444
+- group: testdata.datasource.grafana.app
+  version: v0alpha1
+  host: localhost
+  port: 7445

pkg/services/apiserver/aggregator/examples/manual-test/apiservice.yaml

@@ -1,3 +1,4 @@
+---
 apiVersion: apiregistration.k8s.io/v1
 kind: APIService
 metadata:
@@ -11,4 +12,4 @@ spec:
   service:
     name: example-apiserver
     namespace: grafana
-    port: 7443
+    port: 7443

pkg/services/apiserver/aggregator/examples/manual-test/externalname.yaml

@@ -1,7 +1,8 @@
+---
 apiVersion: service.grafana.app/v0alpha1
 kind: ExternalName
 metadata:
   name: example-apiserver
   namespace: grafana
 spec:
-  host: localhost
+  host: localhost

pkg/services/apiserver/config.go

@@ -41,6 +41,9 @@ func applyGrafanaConfig(cfg *setting.Cfg, features featuremgmt.FeatureToggles, o
 	o.AggregatorOptions.ProxyClientCertFile = apiserverCfg.Key("proxy_client_cert_file").MustString("")
 	o.AggregatorOptions.ProxyClientKeyFile = apiserverCfg.Key("proxy_client_key_file").MustString("")
 
+	o.AggregatorOptions.APIServiceCABundleFile = apiserverCfg.Key("apiservice_ca_bundle_file").MustString("")
+	o.AggregatorOptions.RemoteServicesFile = apiserverCfg.Key("remote_services_file").MustString("")
+
 	o.RecommendedOptions.Admission = nil
 	o.RecommendedOptions.CoreAPI = nil
 

pkg/services/apiserver/options/aggregator.go

@@ -23,9 +23,11 @@ import (
 
 // AggregatorServerOptions contains the state for the aggregator apiserver
 type AggregatorServerOptions struct {
-	AlternateDNS        []string
-	ProxyClientCertFile string
-	ProxyClientKeyFile  string
+	AlternateDNS           []string
+	ProxyClientCertFile    string
+	ProxyClientKeyFile     string
+	RemoteServicesFile     string
+	APIServiceCABundleFile string
 }
 
 func NewAggregatorServerOptions() *AggregatorServerOptions {

pkg/services/apiserver/service.go

@@ -366,12 +366,16 @@ func (s *service) startAggregator(
 	serverConfig *genericapiserver.RecommendedConfig,
 	server *genericapiserver.GenericAPIServer,
 ) (*genericapiserver.GenericAPIServer, error) {
-	aggregatorConfig, aggregatorInformers, err := aggregator.CreateAggregatorConfig(s.options, *serverConfig)
+	externalNamesNamespace := "default"
+	if s.cfg.StackID != "" {
+		externalNamesNamespace = s.cfg.StackID
+	}
+	aggregatorConfig, err := aggregator.CreateAggregatorConfig(s.options, *serverConfig, externalNamesNamespace)
 	if err != nil {
 		return nil, err
 	}
 
-	aggregatorServer, err := aggregator.CreateAggregatorServer(aggregatorConfig, aggregatorInformers, server)
+	aggregatorServer, err := aggregator.CreateAggregatorServer(aggregatorConfig.KubeAggregatorConfig, aggregatorConfig.Informers, aggregatorConfig.RemoteServicesConfig, server)
 	if err != nil {
 		return nil, err
 	}