mirror of https://github.com/grafana/grafana
K8s: Refactor config/options for aggregation (#81739)
Todd Treece
2 years ago
committed by
GitHub
parent
7a17963ab9
commit
67b6be5515
@ -1,628 +0,0 @@ |
||||
// SPDX-License-Identifier: AGPL-3.0-only
|
||||
// Provenance-includes-location: https://github.com/kubernetes/kubernetes/blob/master/cmd/kube-apiserver/app/aggregator.go
|
||||
// Provenance-includes-license: Apache-2.0
|
||||
// Provenance-includes-copyright: The Kubernetes Authors.
|
||||
// Provenance-includes-location: https://github.com/kubernetes/kubernetes/blob/master/cmd/kube-apiserver/app/server.go
|
||||
// Provenance-includes-license: Apache-2.0
|
||||
// Provenance-includes-copyright: The Kubernetes Authors.
|
||||
// Provenance-includes-location: https://github.com/kubernetes/kubernetes/blob/master/pkg/controlplane/apiserver/apiextensions.go
|
||||
// Provenance-includes-license: Apache-2.0
|
||||
// Provenance-includes-copyright: The Kubernetes Authors.
|
||||
|
||||
package aggregator |
||||
|
||||
import ( |
||||
"crypto/tls" |
||||
"fmt" |
||||
"io" |
||||
"net" |
||||
"net/http" |
||||
"path" |
||||
"strings" |
||||
"sync" |
||||
"time" |
||||
|
||||
servicev0alpha1 "github.com/grafana/grafana/pkg/apis/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/registry/apis/service" |
||||
grafanaAPIServer "github.com/grafana/grafana/pkg/services/grafana-apiserver" |
||||
filestorage "github.com/grafana/grafana/pkg/services/grafana-apiserver/storage/file" |
||||
|
||||
apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1" |
||||
apiextensionsapiserver "k8s.io/apiextensions-apiserver/pkg/apiserver" |
||||
apiextensionsinformers "k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions" |
||||
apiextensionsopenapi "k8s.io/apiextensions-apiserver/pkg/generated/openapi" |
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" |
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" |
||||
"k8s.io/apimachinery/pkg/runtime" |
||||
"k8s.io/apimachinery/pkg/runtime/schema" |
||||
"k8s.io/apimachinery/pkg/runtime/serializer" |
||||
utilnet "k8s.io/apimachinery/pkg/util/net" |
||||
"k8s.io/apimachinery/pkg/util/sets" |
||||
"k8s.io/apiserver/pkg/endpoints/discovery/aggregated" |
||||
openapinamer "k8s.io/apiserver/pkg/endpoints/openapi" |
||||
genericfeatures "k8s.io/apiserver/pkg/features" |
||||
genericapiserver "k8s.io/apiserver/pkg/server" |
||||
"k8s.io/apiserver/pkg/server/healthz" |
||||
"k8s.io/apiserver/pkg/server/options" |
||||
"k8s.io/apiserver/pkg/server/resourceconfig" |
||||
utilfeature "k8s.io/apiserver/pkg/util/feature" |
||||
"k8s.io/apiserver/pkg/util/openapi" |
||||
"k8s.io/client-go/informers" |
||||
"k8s.io/client-go/kubernetes/fake" |
||||
"k8s.io/client-go/tools/cache" |
||||
"k8s.io/klog/v2" |
||||
v1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1" |
||||
v1helper "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1/helper" |
||||
apiregistrationv1beta1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1beta1" |
||||
aggregatorapiserver "k8s.io/kube-aggregator/pkg/apiserver" |
||||
aggregatorscheme "k8s.io/kube-aggregator/pkg/apiserver/scheme" |
||||
apiregistrationclientset "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset" |
||||
apiregistrationclient "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset/typed/apiregistration/v1" |
||||
apiregistrationInformers "k8s.io/kube-aggregator/pkg/client/informers/externalversions/apiregistration/v1" |
||||
"k8s.io/kube-aggregator/pkg/controllers/autoregister" |
||||
apiserver "k8s.io/kube-aggregator/pkg/controllers/status" |
||||
aggregatoropenapi "k8s.io/kube-aggregator/pkg/generated/openapi" |
||||
"k8s.io/kube-openapi/pkg/common" |
||||
) |
||||
|
||||
// AggregatorServerOptions contains the state for the aggregator apiserver
|
||||
type AggregatorServerOptions struct { |
||||
Builders []grafanaAPIServer.APIGroupBuilder |
||||
AlternateDNS []string |
||||
Config *Config |
||||
serviceResolver ServiceResolver |
||||
|
||||
sharedInformerFactory informersv0alpha1.SharedInformerFactory |
||||
|
||||
StdOut io.Writer |
||||
StdErr io.Writer |
||||
} |
||||
|
||||
func NewAggregatorServerOptions(out, errOut io.Writer, |
||||
options *options.RecommendedOptions, |
||||
extraConfig *ExtraConfig, |
||||
) (*AggregatorServerOptions, error) { |
||||
sharedConfig, err := initSharedConfig(options, aggregatorscheme.Codecs, nil) |
||||
if err != nil { |
||||
klog.Errorf("Error creating shared config: %s", err) |
||||
return nil, err |
||||
} |
||||
|
||||
sharedInformerFactory, err := initSharedInformerFactory(sharedConfig) |
||||
if err != nil { |
||||
klog.Errorf("Error creating shared informer factory: %s", err) |
||||
return nil, err |
||||
} |
||||
|
||||
serviceResolver, err := initServiceResolver(sharedInformerFactory) |
||||
if err != nil { |
||||
klog.Errorf("Error creating service resolver: %s", err) |
||||
return nil, err |
||||
} |
||||
|
||||
fakeInformers := informers.NewSharedInformerFactory(fake.NewSimpleClientset(), 10*time.Minute) |
||||
builders := []grafanaAPIServer.APIGroupBuilder{ |
||||
service.NewServiceAPIBuilder(), |
||||
} |
||||
|
||||
extensionsConfig, err := initApiExtensionsConfig(options, sharedConfig, fakeInformers, serviceResolver, extraConfig.DataPath) |
||||
if err != nil { |
||||
klog.Errorf("Error creating extensions config: %s", err) |
||||
return nil, err |
||||
} |
||||
|
||||
aggregatorConfig, err := initAggregatorConfig(options, sharedConfig, extraConfig, fakeInformers, builders, serviceResolver, extraConfig.DataPath) |
||||
if err != nil { |
||||
klog.Errorf("Error creating aggregator config: %s", err) |
||||
return nil, err |
||||
} |
||||
|
||||
return &AggregatorServerOptions{ |
||||
StdOut: out, |
||||
StdErr: errOut, |
||||
Builders: builders, |
||||
sharedInformerFactory: sharedInformerFactory, |
||||
serviceResolver: serviceResolver, |
||||
Config: &Config{ |
||||
Aggregator: aggregatorConfig, |
||||
ApiExtensions: extensionsConfig, |
||||
|
||||
SharedConfig: sharedConfig, |
||||
extraConfig: extraConfig, |
||||
}, |
||||
}, nil |
||||
} |
||||
|
||||
func (o *AggregatorServerOptions) LoadAPIGroupBuilders() error { |
||||
// Install schemas
|
||||
for _, b := range o.Builders { |
||||
if err := b.InstallSchema(aggregatorscheme.Scheme); err != nil { |
||||
return err |
||||
} |
||||
} |
||||
return nil |
||||
} |
||||
|
||||
func initSharedConfig(options *options.RecommendedOptions, codecs serializer.CodecFactory, alternateDNS []string) (*genericapiserver.RecommendedConfig, error) { |
||||
if err := options.SecureServing.MaybeDefaultWithSelfSignedCerts( |
||||
"localhost", alternateDNS, []net.IP{net.IPv4(127, 0, 0, 1)}, |
||||
); err != nil { |
||||
return nil, fmt.Errorf("error creating self-signed certificates: %v", err) |
||||
} |
||||
|
||||
options.Authentication.RemoteKubeConfigFileOptional = true |
||||
options.Authorization.RemoteKubeConfigFileOptional = true |
||||
|
||||
options.Admission = nil |
||||
|
||||
if options.CoreAPI.CoreAPIKubeconfigPath == "" { |
||||
options.CoreAPI = nil |
||||
} |
||||
|
||||
serverConfig := genericapiserver.NewRecommendedConfig(codecs) |
||||
|
||||
// NOTE: AggregatedDiscoveryGroupManager in kube-apiserver is set up by controlplane APIServerConfig creation
|
||||
// Here, we adopt that one line in addition to what recommendedOptions gives us
|
||||
// Without it, CRDs work on API routes (and are registered in openapi) but not discoverable by kubectl
|
||||
serverConfig.AggregatedDiscoveryGroupManager = aggregated.NewResourceManager("apis") |
||||
|
||||
if options.CoreAPI == nil { |
||||
if err := modifiedApplyTo(options, serverConfig); err != nil { |
||||
return nil, err |
||||
} |
||||
} else { |
||||
if err := options.ApplyTo(serverConfig); err != nil { |
||||
return nil, err |
||||
} |
||||
} |
||||
|
||||
return serverConfig, nil |
||||
} |
||||
|
||||
// A copy of ApplyTo in recommended.go, but for >= 0.28, server pkg in apiserver does a bit extra causing
|
||||
// a panic when CoreAPI is set to nil
|
||||
func modifiedApplyTo(options *options.RecommendedOptions, config *genericapiserver.RecommendedConfig) error { |
||||
if err := options.Etcd.ApplyTo(&config.Config); err != nil { |
||||
return err |
||||
} |
||||
if err := options.EgressSelector.ApplyTo(&config.Config); err != nil { |
||||
return err |
||||
} |
||||
if err := options.Traces.ApplyTo(config.Config.EgressSelector, &config.Config); err != nil { |
||||
return err |
||||
} |
||||
if err := options.SecureServing.ApplyTo(&config.Config.SecureServing, &config.Config.LoopbackClientConfig); err != nil { |
||||
return err |
||||
} |
||||
if err := options.Authentication.ApplyTo(&config.Config.Authentication, config.SecureServing, config.OpenAPIConfig); err != nil { |
||||
return err |
||||
} |
||||
if err := options.Authorization.ApplyTo(&config.Config.Authorization); err != nil { |
||||
return err |
||||
} |
||||
if err := options.Audit.ApplyTo(&config.Config); err != nil { |
||||
return err |
||||
} |
||||
|
||||
// TODO: determine whether we need flow control (API priority and fairness)
|
||||
//if err := options.Features.ApplyTo(&config.Config); err != nil {
|
||||
// return err
|
||||
//}
|
||||
|
||||
if err := options.CoreAPI.ApplyTo(config); err != nil { |
||||
return err |
||||
} |
||||
|
||||
_, err := options.ExtraAdmissionInitializers(config) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
return nil |
||||
} |
||||
|
||||
func getMergedOpenAPIDefinitions(builders []grafanaAPIServer.APIGroupBuilder, ref common.ReferenceCallback) map[string]common.OpenAPIDefinition { |
||||
// Add OpenAPI specs for each group+version
|
||||
prerequisiteAPIs := grafanaAPIServer.GetOpenAPIDefinitions(builders)(ref) |
||||
aggregatorAPIs := aggregatoropenapi.GetOpenAPIDefinitions(ref) |
||||
|
||||
for k, v := range prerequisiteAPIs { |
||||
aggregatorAPIs[k] = v |
||||
} |
||||
|
||||
return aggregatorAPIs |
||||
} |
||||
|
||||
func initSharedInformerFactory(sharedConfig *genericapiserver.RecommendedConfig) (informersv0alpha1.SharedInformerFactory, error) { |
||||
serviceClient, err := serviceclientset.NewForConfig(sharedConfig.LoopbackClientConfig) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
return informersv0alpha1.NewSharedInformerFactory( |
||||
serviceClient, |
||||
5*time.Minute, // this is effectively used as a refresh interval right now. Might want to do something nicer later on.
|
||||
), nil |
||||
} |
||||
|
||||
func initServiceResolver(factory informersv0alpha1.SharedInformerFactory) (apiserver.ServiceResolver, error) { |
||||
return NewExternalNameResolver(factory.Service().V0alpha1().ExternalNames().Lister()), nil |
||||
} |
||||
|
||||
func initApiExtensionsConfig(options *options.RecommendedOptions, |
||||
sharedConfig *genericapiserver.RecommendedConfig, |
||||
fakeInfomers informers.SharedInformerFactory, |
||||
serviceResolver apiserver.ServiceResolver, |
||||
dataPath string, |
||||
) (*apiextensionsapiserver.Config, error) { |
||||
// make a shallow copy to let us twiddle a few things
|
||||
// most of the config actually remains the same. We only need to mess with a couple items related to the particulars of the api extensions
|
||||
genericConfig := sharedConfig.Config |
||||
|
||||
genericConfig.PostStartHooks = map[string]genericapiserver.PostStartHookConfigEntry{} |
||||
genericConfig.RESTOptionsGetter = nil |
||||
|
||||
// copy the etcd options so we don't mutate originals.
|
||||
// we assume that the etcd options have been completed already. avoid messing with anything outside
|
||||
// of changes to StorageConfig as that may lead to unexpected behavior when the options are applied.
|
||||
etcdOptions := *options.Etcd |
||||
// this is where the true decodable levels come from.
|
||||
etcdOptions.StorageConfig.Codec = apiextensionsapiserver.Codecs.LegacyCodec(apiextensionsv1beta1.SchemeGroupVersion, v1.SchemeGroupVersion) |
||||
// prefer the more compact serialization (v1beta1) for storage until https://issue.k8s.io/82292 is resolved for objects whose v1 serialization is too big but whose v1beta1 serialization can be stored
|
||||
etcdOptions.StorageConfig.EncodeVersioner = runtime.NewMultiGroupVersioner(apiextensionsv1beta1.SchemeGroupVersion, schema.GroupKind{Group: apiextensionsv1beta1.GroupName}) |
||||
etcdOptions.SkipHealthEndpoints = true // avoid double wiring of health checks
|
||||
if err := etcdOptions.ApplyTo(&genericConfig); err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
restOptionsGetter := filestorage.NewRESTOptionsGetter(path.Join(dataPath, "grafana-apiextensionsserver"), etcdOptions.StorageConfig) |
||||
genericConfig.RESTOptionsGetter = restOptionsGetter |
||||
|
||||
// NOTE: ignoring genericConfig.ResourceTransformers in crdOptionsGetter creation for now
|
||||
// crdOptionsGetter := apiextensionsoptions.NewCRDRESTOptionsGetter(etcdOptions, genericConfig.ResourceTransformers, )
|
||||
// The following is equivalent code to apiextensionsoptions.NewCRDRESTOptionsGetter with lesser dependencies
|
||||
crdEtcdOptions := etcdOptions |
||||
crdEtcdOptions.StorageConfig.Codec = unstructured.UnstructuredJSONScheme |
||||
crdEtcdOptions.StorageConfig.StorageObjectCountTracker = genericConfig.StorageObjectCountTracker |
||||
crdEtcdOptions.WatchCacheSizes = nil // this control is not provided for custom resources
|
||||
|
||||
// override MergedResourceConfig with apiextensions defaults and registry
|
||||
mergedResourceConfig, err := resourceconfig.MergeAPIResourceConfigs(apiextensionsapiserver.DefaultAPIResourceConfigSource(), nil, apiextensionsapiserver.Scheme) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
genericConfig.MergedResourceConfig = mergedResourceConfig |
||||
|
||||
genericConfig.OpenAPIV3Config = genericapiserver.DefaultOpenAPIV3Config(openapi.GetOpenAPIDefinitionsWithoutDisabledFeatures(apiextensionsopenapi.GetOpenAPIDefinitions), openapinamer.NewDefinitionNamer(apiextensionsapiserver.Scheme, apiextensionsapiserver.Scheme)) |
||||
|
||||
apiextensionsConfig := &apiextensionsapiserver.Config{ |
||||
GenericConfig: &genericapiserver.RecommendedConfig{ |
||||
Config: genericConfig, |
||||
SharedInformerFactory: fakeInfomers, |
||||
}, |
||||
ExtraConfig: apiextensionsapiserver.ExtraConfig{ |
||||
CRDRESTOptionsGetter: filestorage.NewRESTOptionsGetter(path.Join(dataPath, "grafana-apiextensionsserver"), crdEtcdOptions.StorageConfig), |
||||
// TODO: remove the hardcod when HA story is more developed
|
||||
MasterCount: 1, |
||||
// TODO: leaving AuthResolverWrapper unset doesn't impact basic operation of CRDs
|
||||
// AuthResolverWrapper: authResolverWrapper,
|
||||
ServiceResolver: serviceResolver, |
||||
}, |
||||
} |
||||
|
||||
// we need to clear the poststarthooks so we don't add them multiple times to all the servers (that fails)
|
||||
apiextensionsConfig.GenericConfig.PostStartHooks = map[string]genericapiserver.PostStartHookConfigEntry{} |
||||
|
||||
return apiextensionsConfig, nil |
||||
} |
||||
|
||||
func initAggregatorConfig(options *options.RecommendedOptions, |
||||
sharedConfig *genericapiserver.RecommendedConfig, |
||||
extra *ExtraConfig, |
||||
fakeInformers informers.SharedInformerFactory, |
||||
builders []grafanaAPIServer.APIGroupBuilder, |
||||
serviceResolver apiserver.ServiceResolver, |
||||
dataPath string, |
||||
) (*aggregatorapiserver.Config, error) { |
||||
// make a shallow copy to let us twiddle a few things
|
||||
// most of the config actually remains the same. We only need to mess with a couple items related to the particulars of the aggregator
|
||||
genericConfig := sharedConfig.Config |
||||
|
||||
genericConfig.PostStartHooks = map[string]genericapiserver.PostStartHookConfigEntry{} |
||||
genericConfig.RESTOptionsGetter = nil |
||||
// prevent generic API server from installing the OpenAPI handler. Aggregator server
|
||||
// has its own customized OpenAPI handler.
|
||||
genericConfig.SkipOpenAPIInstallation = true |
||||
mergedResourceConfig, err := resourceconfig.MergeAPIResourceConfigs(aggregatorapiserver.DefaultAPIResourceConfigSource(), nil, aggregatorscheme.Scheme) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
genericConfig.MergedResourceConfig = mergedResourceConfig |
||||
|
||||
getOpenAPIDefinitionsFunc := func() common.GetOpenAPIDefinitions { |
||||
return func(ref common.ReferenceCallback) map[string]common.OpenAPIDefinition { |
||||
return getMergedOpenAPIDefinitions(builders, ref) |
||||
} |
||||
} |
||||
|
||||
namer := openapinamer.NewDefinitionNamer(aggregatorscheme.Scheme, apiextensionsapiserver.Scheme) |
||||
genericConfig.OpenAPIV3Config = genericapiserver.DefaultOpenAPIV3Config(getOpenAPIDefinitionsFunc(), namer) |
||||
genericConfig.OpenAPIV3Config.Info.Title = "Kubernetes" |
||||
genericConfig.OpenAPIConfig = genericapiserver.DefaultOpenAPIConfig(getOpenAPIDefinitionsFunc(), namer) |
||||
genericConfig.OpenAPIConfig.Info.Title = "Kubernetes" |
||||
|
||||
if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.StorageVersionAPI) && |
||||
utilfeature.DefaultFeatureGate.Enabled(genericfeatures.APIServerIdentity) { |
||||
// Add StorageVersionPrecondition handler to aggregator-apiserver.
|
||||
// The handler will block write requests to built-in resources until the
|
||||
// target resources' storage versions are up-to-date.
|
||||
genericConfig.BuildHandlerChainFunc = genericapiserver.BuildHandlerChainWithStorageVersionPrecondition |
||||
} |
||||
|
||||
// copy the etcd options so we don't mutate originals.
|
||||
// we assume that the etcd options have been completed already. avoid messing with anything outside
|
||||
// of changes to StorageConfig as that may lead to unexpected behavior when the options are applied.
|
||||
etcdOptions := *options.Etcd |
||||
etcdOptions.StorageConfig.Codec = aggregatorscheme.Codecs.LegacyCodec(v1.SchemeGroupVersion, |
||||
apiregistrationv1beta1.SchemeGroupVersion, |
||||
servicev0alpha1.SchemeGroupVersion) |
||||
etcdOptions.StorageConfig.EncodeVersioner = runtime.NewMultiGroupVersioner(v1.SchemeGroupVersion, |
||||
schema.GroupKind{Group: apiregistrationv1beta1.GroupName}, |
||||
schema.GroupKind{Group: servicev0alpha1.GROUP}) |
||||
etcdOptions.SkipHealthEndpoints = true // avoid double wiring of health checks
|
||||
if err := etcdOptions.ApplyTo(&genericConfig); err != nil { |
||||
return nil, err |
||||
} |
||||
genericConfig.RESTOptionsGetter = filestorage.NewRESTOptionsGetter(path.Join(dataPath, "grafana-aggregator"), etcdOptions.StorageConfig) |
||||
|
||||
genericConfig.DisabledPostStartHooks = genericConfig.DisabledPostStartHooks.Insert("apiservice-status-available-controller") |
||||
genericConfig.DisabledPostStartHooks = genericConfig.DisabledPostStartHooks.Insert("start-kube-aggregator-informers") |
||||
|
||||
aggregatorConfig := &aggregatorapiserver.Config{ |
||||
GenericConfig: &genericapiserver.RecommendedConfig{ |
||||
Config: genericConfig, |
||||
SharedInformerFactory: fakeInformers, |
||||
ClientConfig: genericConfig.LoopbackClientConfig, |
||||
}, |
||||
ExtraConfig: aggregatorapiserver.ExtraConfig{ |
||||
ProxyClientCertFile: extra.ProxyClientCertFile, |
||||
ProxyClientKeyFile: extra.ProxyClientKeyFile, |
||||
// NOTE: while ProxyTransport can be skipped in the configuration, it allows honoring
|
||||
// DISABLE_HTTP2, HTTPS_PROXY and NO_PROXY env vars as needed
|
||||
ProxyTransport: createProxyTransport(), |
||||
}, |
||||
} |
||||
|
||||
aggregatorConfig.ExtraConfig.ServiceResolver = serviceResolver |
||||
|
||||
// we need to clear the poststarthooks so we don't add them multiple times to all the servers (that fails)
|
||||
aggregatorConfig.GenericConfig.PostStartHooks = map[string]genericapiserver.PostStartHookConfigEntry{} |
||||
|
||||
return aggregatorConfig, nil |
||||
} |
||||
|
||||
func (o *AggregatorServerOptions) CreateAggregatorServer(delegateAPIServer genericapiserver.DelegationTarget, apiExtensionsInformers apiextensionsinformers.SharedInformerFactory) (*aggregatorapiserver.APIAggregator, error) { |
||||
completedConfig := o.Config.AggregatorComplete |
||||
aggregatorServer, err := completedConfig.NewWithDelegate(delegateAPIServer) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
// create controllers for auto-registration
|
||||
apiRegistrationClient, err := apiregistrationclient.NewForConfig(completedConfig.GenericConfig.LoopbackClientConfig) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
autoRegistrationController := autoregister.NewAutoRegisterController(aggregatorServer.APIRegistrationInformers.Apiregistration().V1().APIServices(), apiRegistrationClient) |
||||
apiServices := apiServicesToRegister(delegateAPIServer, autoRegistrationController) |
||||
|
||||
crdRegistrationController := NewCRDRegistrationController( |
||||
apiExtensionsInformers.Apiextensions().V1().CustomResourceDefinitions(), |
||||
autoRegistrationController) |
||||
|
||||
// Imbue all builtin group-priorities onto the aggregated discovery
|
||||
if completedConfig.GenericConfig.AggregatedDiscoveryGroupManager != nil { |
||||
for gv, entry := range apiVersionPriorities { |
||||
completedConfig.GenericConfig.AggregatedDiscoveryGroupManager.SetGroupVersionPriority(metav1.GroupVersion(gv), int(entry.group), int(entry.version)) |
||||
} |
||||
} |
||||
|
||||
err = aggregatorServer.GenericAPIServer.AddPostStartHook("kube-apiserver-autoregistration", func(context genericapiserver.PostStartHookContext) error { |
||||
go crdRegistrationController.Run(5, context.StopCh) |
||||
go func() { |
||||
crdRegistrationController.WaitForInitialSync() |
||||
autoRegistrationController.Run(5, context.StopCh) |
||||
}() |
||||
return nil |
||||
}) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
err = aggregatorServer.GenericAPIServer.AddBootSequenceHealthChecks( |
||||
makeAPIServiceAvailableHealthCheck( |
||||
"autoregister-completion", |
||||
apiServices, |
||||
aggregatorServer.APIRegistrationInformers.Apiregistration().V1().APIServices(), |
||||
), |
||||
) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
apiregistrationClient, err := apiregistrationclientset.NewForConfig(completedConfig.GenericConfig.LoopbackClientConfig) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
availableController, err := NewAvailableConditionController( |
||||
aggregatorServer.APIRegistrationInformers.Apiregistration().V1().APIServices(), |
||||
o.sharedInformerFactory.Service().V0alpha1().ExternalNames(), |
||||
apiregistrationClient.ApiregistrationV1(), |
||||
nil, |
||||
(func() ([]byte, []byte))(nil), |
||||
completedConfig.ExtraConfig.ServiceResolver, |
||||
) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
aggregatorServer.GenericAPIServer.AddPostStartHookOrDie("apiservice-status-override-available-controller", func(context genericapiserver.PostStartHookContext) error { |
||||
// if we end up blocking for long periods of time, we may need to increase workers.
|
||||
go availableController.Run(5, context.StopCh) |
||||
return nil |
||||
}) |
||||
|
||||
aggregatorServer.GenericAPIServer.AddPostStartHookOrDie("start-grafana-aggregator-informers", func(context genericapiserver.PostStartHookContext) error { |
||||
o.sharedInformerFactory.Start(context.StopCh) |
||||
aggregatorServer.APIRegistrationInformers.Start(context.StopCh) |
||||
return nil |
||||
}) |
||||
|
||||
// Install the API Group+version
|
||||
for _, b := range o.Builders { |
||||
g, err := b.GetAPIGroupInfo(aggregatorscheme.Scheme, aggregatorscheme.Codecs, completedConfig.GenericConfig.RESTOptionsGetter) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
if g == nil || len(g.PrioritizedVersions) < 1 { |
||||
continue |
||||
} |
||||
err = aggregatorServer.GenericAPIServer.InstallAPIGroup(g) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
} |
||||
|
||||
return aggregatorServer, nil |
||||
} |
||||
|
||||
func makeAPIService(gv schema.GroupVersion) *v1.APIService { |
||||
apiServicePriority, ok := apiVersionPriorities[gv] |
||||
if !ok { |
||||
// if we aren't found, then we shouldn't register ourselves because it could result in a CRD group version
|
||||
// being permanently stuck in the APIServices list.
|
||||
klog.Infof("Skipping APIService creation for %v", gv) |
||||
return nil |
||||
} |
||||
return &v1.APIService{ |
||||
ObjectMeta: metav1.ObjectMeta{Name: gv.Version + "." + gv.Group}, |
||||
Spec: v1.APIServiceSpec{ |
||||
Group: gv.Group, |
||||
Version: gv.Version, |
||||
GroupPriorityMinimum: apiServicePriority.group, |
||||
VersionPriority: apiServicePriority.version, |
||||
}, |
||||
} |
||||
} |
||||
|
||||
// makeAPIServiceAvailableHealthCheck returns a healthz check that returns healthy
|
||||
// once all of the specified services have been observed to be available at least once.
|
||||
func makeAPIServiceAvailableHealthCheck(name string, apiServices []*v1.APIService, apiServiceInformer apiregistrationInformers.APIServiceInformer) healthz.HealthChecker { |
||||
// Track the auto-registered API services that have not been observed to be available yet
|
||||
pendingServiceNamesLock := &sync.RWMutex{} |
||||
pendingServiceNames := sets.NewString() |
||||
for _, service := range apiServices { |
||||
pendingServiceNames.Insert(service.Name) |
||||
} |
||||
|
||||
// When an APIService in the list is seen as available, remove it from the pending list
|
||||
handleAPIServiceChange := func(service *v1.APIService) { |
||||
pendingServiceNamesLock.Lock() |
||||
defer pendingServiceNamesLock.Unlock() |
||||
if !pendingServiceNames.Has(service.Name) { |
||||
return |
||||
} |
||||
if v1helper.IsAPIServiceConditionTrue(service, v1.Available) { |
||||
pendingServiceNames.Delete(service.Name) |
||||
} |
||||
} |
||||
|
||||
// Watch add/update events for APIServices
|
||||
_, _ = apiServiceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ |
||||
AddFunc: func(obj interface{}) { handleAPIServiceChange(obj.(*v1.APIService)) }, |
||||
UpdateFunc: func(old, new interface{}) { handleAPIServiceChange(new.(*v1.APIService)) }, |
||||
}) |
||||
|
||||
// Don't return healthy until the pending list is empty
|
||||
return healthz.NamedCheck(name, func(r *http.Request) error { |
||||
pendingServiceNamesLock.RLock() |
||||
defer pendingServiceNamesLock.RUnlock() |
||||
if pendingServiceNames.Len() > 0 { |
||||
return fmt.Errorf("missing APIService: %v", pendingServiceNames.List()) |
||||
} |
||||
return nil |
||||
}) |
||||
} |
||||
|
||||
// priority defines group priority that is used in discovery. This controls
|
||||
// group position in the kubectl output.
|
||||
type priority struct { |
||||
// group indicates the order of the group relative to other groups.
|
||||
group int32 |
||||
// version indicates the relative order of the version inside of its group.
|
||||
version int32 |
||||
} |
||||
|
||||
// The proper way to resolve this letting the aggregator know the desired group and version-within-group order of the underlying servers
|
||||
// is to refactor the genericapiserver.DelegationTarget to include a list of priorities based on which APIs were installed.
|
||||
// This requires the APIGroupInfo struct to evolve and include the concept of priorities and to avoid mistakes, the core storage map there needs to be updated.
|
||||
// That ripples out every bit as far as you'd expect, so for 1.7 we'll include the list here instead of being built up during storage.
|
||||
var apiVersionPriorities = map[schema.GroupVersion]priority{ |
||||
{Group: "", Version: "v1"}: {group: 18000, version: 1}, |
||||
// to my knowledge, nothing below here collides
|
||||
{Group: "admissionregistration.k8s.io", Version: "v1"}: {group: 16700, version: 15}, |
||||
{Group: "admissionregistration.k8s.io", Version: "v1beta1"}: {group: 16700, version: 12}, |
||||
{Group: "admissionregistration.k8s.io", Version: "v1alpha1"}: {group: 16700, version: 9}, |
||||
{Group: "apiextensions.k8s.io", Version: "v1"}: {group: 16700, version: 15}, |
||||
// Append a new group to the end of the list if unsure.
|
||||
// You can use min(existing group)-100 as the initial value for a group.
|
||||
// Version can be set to 9 (to have space around) for a new group.
|
||||
} |
||||
|
||||
func apiServicesToRegister(delegateAPIServer genericapiserver.DelegationTarget, registration autoregister.AutoAPIServiceRegistration) []*v1.APIService { |
||||
apiServices := []*v1.APIService{} |
||||
|
||||
for _, curr := range delegateAPIServer.ListedPaths() { |
||||
if curr == "/api/v1" { |
||||
apiService := makeAPIService(schema.GroupVersion{Group: "", Version: "v1"}) |
||||
registration.AddAPIServiceToSyncOnStart(apiService) |
||||
apiServices = append(apiServices, apiService) |
||||
continue |
||||
} |
||||
|
||||
if !strings.HasPrefix(curr, "/apis/") { |
||||
continue |
||||
} |
||||
// this comes back in a list that looks like /apis/rbac.authorization.k8s.io/v1alpha1
|
||||
tokens := strings.Split(curr, "/") |
||||
if len(tokens) != 4 { |
||||
continue |
||||
} |
||||
|
||||
apiService := makeAPIService(schema.GroupVersion{Group: tokens[2], Version: tokens[3]}) |
||||
if apiService == nil { |
||||
continue |
||||
} |
||||
registration.AddAPIServiceToSyncOnStart(apiService) |
||||
apiServices = append(apiServices, apiService) |
||||
} |
||||
|
||||
return apiServices |
||||
} |
||||
|
||||
// NOTE: below function imported from https://github.com/kubernetes/kubernetes/blob/master/cmd/kube-apiserver/app/server.go#L197
|
||||
// createProxyTransport creates the dialer infrastructure to connect to the api servers.
|
||||
func createProxyTransport() *http.Transport { |
||||
// NOTE: We don't set proxyDialerFn but the below SetTransportDefaults will
|
||||
// See https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apimachinery/pkg/util/net/http.go#L109
|
||||
var proxyDialerFn utilnet.DialFunc |
||||
// Proxying to services is IP-based... don't expect to be able to verify the hostname
|
||||
proxyTLSClientConfig := &tls.Config{InsecureSkipVerify: true} |
||||
proxyTransport := utilnet.SetTransportDefaults(&http.Transport{ |
||||
DialContext: proxyDialerFn, |
||||
TLSClientConfig: proxyTLSClientConfig, |
||||
}) |
||||
return proxyTransport |
||||
} |
||||
@ -1,57 +0,0 @@ |
||||
package aggregator |
||||
|
||||
import ( |
||||
"github.com/spf13/pflag" |
||||
apiextensionsapiserver "k8s.io/apiextensions-apiserver/pkg/apiserver" |
||||
genericapiserver "k8s.io/apiserver/pkg/server" |
||||
"k8s.io/apiserver/pkg/server/options" |
||||
aggregatorapiserver "k8s.io/kube-aggregator/pkg/apiserver" |
||||
) |
||||
|
||||
type ExtraConfig struct { |
||||
ProxyClientCertFile string |
||||
ProxyClientKeyFile string |
||||
|
||||
DataPath string |
||||
} |
||||
|
||||
type Config struct { |
||||
Aggregator *aggregatorapiserver.Config |
||||
ApiExtensions *apiextensionsapiserver.Config |
||||
|
||||
AggregatorComplete aggregatorapiserver.CompletedConfig |
||||
ApiExtensionsComplete apiextensionsapiserver.CompletedConfig |
||||
|
||||
recommendedOptions *options.RecommendedOptions |
||||
SharedConfig *genericapiserver.RecommendedConfig |
||||
extraConfig *ExtraConfig |
||||
} |
||||
|
||||
func (c *Config) AddFlags(fs *pflag.FlagSet) { |
||||
if c == nil { |
||||
return |
||||
} |
||||
|
||||
c.recommendedOptions.AddFlags(fs) |
||||
} |
||||
|
||||
func (c *Config) Complete() { |
||||
if c == nil { |
||||
return |
||||
} |
||||
|
||||
c.ApiExtensionsComplete = c.ApiExtensions.Complete() |
||||
c.AggregatorComplete = c.Aggregator.Complete() |
||||
} |
||||
|
||||
func (ec *ExtraConfig) AddFlags(fs *pflag.FlagSet) { |
||||
if ec == nil { |
||||
return |
||||
} |
||||
|
||||
fs.StringVar(&ec.ProxyClientCertFile, "proxy-client-cert-file", ec.ProxyClientCertFile, |
||||
"path to proxy client cert file") |
||||
|
||||
fs.StringVar(&ec.ProxyClientKeyFile, "proxy-client-key-file", ec.ProxyClientKeyFile, |
||||
"path to proxy client cert file") |
||||
} |
||||
@ -1,214 +0,0 @@ |
||||
// SPDX-License-Identifier: AGPL-3.0-only
|
||||
// Provenance-includes-location: https://github.com/kubernetes/kubernetes/blob/master/pkg/controlplane/controller/crdregistration/crdregistration_controller.go
|
||||
// Provenance-includes-license: Apache-2.0
|
||||
// Provenance-includes-copyright: The Kubernetes Authors.
|
||||
|
||||
package aggregator |
||||
|
||||
import ( |
||||
"fmt" |
||||
"time" |
||||
|
||||
"k8s.io/klog/v2" |
||||
|
||||
apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" |
||||
crdinformers "k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions/apiextensions/v1" |
||||
crdlisters "k8s.io/apiextensions-apiserver/pkg/client/listers/apiextensions/v1" |
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" |
||||
"k8s.io/apimachinery/pkg/labels" |
||||
"k8s.io/apimachinery/pkg/runtime/schema" |
||||
utilruntime "k8s.io/apimachinery/pkg/util/runtime" |
||||
"k8s.io/apimachinery/pkg/util/wait" |
||||
"k8s.io/client-go/tools/cache" |
||||
"k8s.io/client-go/util/workqueue" |
||||
v1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1" |
||||
) |
||||
|
||||
// AutoAPIServiceRegistration is an interface which callers can re-declare locally and properly cast to for
|
||||
// adding and removing APIServices
|
||||
type AutoAPIServiceRegistration interface { |
||||
// AddAPIServiceToSync adds an API service to auto-register.
|
||||
AddAPIServiceToSync(in *v1.APIService) |
||||
// RemoveAPIServiceToSync removes an API service to auto-register.
|
||||
RemoveAPIServiceToSync(name string) |
||||
} |
||||
|
||||
type crdRegistrationController struct { |
||||
crdLister crdlisters.CustomResourceDefinitionLister |
||||
crdSynced cache.InformerSynced |
||||
|
||||
apiServiceRegistration AutoAPIServiceRegistration |
||||
|
||||
syncHandler func(groupVersion schema.GroupVersion) error |
||||
|
||||
syncedInitialSet chan struct{} |
||||
|
||||
// queue is where incoming work is placed to de-dup and to allow "easy" rate limited requeues on errors
|
||||
// this is actually keyed by a groupVersion
|
||||
queue workqueue.RateLimitingInterface |
||||
} |
||||
|
||||
// NewCRDRegistrationController returns a controller which will register CRD GroupVersions with the auto APIService registration
|
||||
// controller so they automatically stay in sync.
|
||||
func NewCRDRegistrationController(crdinformer crdinformers.CustomResourceDefinitionInformer, apiServiceRegistration AutoAPIServiceRegistration) *crdRegistrationController { |
||||
c := &crdRegistrationController{ |
||||
crdLister: crdinformer.Lister(), |
||||
crdSynced: crdinformer.Informer().HasSynced, |
||||
apiServiceRegistration: apiServiceRegistration, |
||||
syncedInitialSet: make(chan struct{}), |
||||
queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "crd_autoregistration_controller"), |
||||
} |
||||
c.syncHandler = c.handleVersionUpdate |
||||
|
||||
_, _ = crdinformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ |
||||
AddFunc: func(obj interface{}) { |
||||
cast := obj.(*apiextensionsv1.CustomResourceDefinition) |
||||
c.enqueueCRD(cast) |
||||
}, |
||||
UpdateFunc: func(oldObj, newObj interface{}) { |
||||
// Enqueue both old and new object to make sure we remove and add appropriate API services.
|
||||
// The working queue will resolve any duplicates and only changes will stay in the queue.
|
||||
c.enqueueCRD(oldObj.(*apiextensionsv1.CustomResourceDefinition)) |
||||
c.enqueueCRD(newObj.(*apiextensionsv1.CustomResourceDefinition)) |
||||
}, |
||||
DeleteFunc: func(obj interface{}) { |
||||
cast, ok := obj.(*apiextensionsv1.CustomResourceDefinition) |
||||
if !ok { |
||||
tombstone, ok := obj.(cache.DeletedFinalStateUnknown) |
||||
if !ok { |
||||
klog.V(2).Infof("Couldn't get object from tombstone %#v", obj) |
||||
return |
||||
} |
||||
cast, ok = tombstone.Obj.(*apiextensionsv1.CustomResourceDefinition) |
||||
if !ok { |
||||
klog.V(2).Infof("Tombstone contained unexpected object: %#v", obj) |
||||
return |
||||
} |
||||
} |
||||
c.enqueueCRD(cast) |
||||
}, |
||||
}) |
||||
|
||||
return c |
||||
} |
||||
|
||||
func (c *crdRegistrationController) Run(workers int, stopCh <-chan struct{}) { |
||||
defer utilruntime.HandleCrash() |
||||
// make sure the work queue is shutdown which will trigger workers to end
|
||||
defer c.queue.ShutDown() |
||||
|
||||
klog.Infof("Starting crd-autoregister controller") |
||||
defer klog.Infof("Shutting down crd-autoregister controller") |
||||
|
||||
// wait for your secondary caches to fill before starting your work
|
||||
if !cache.WaitForNamedCacheSync("crd-autoregister", stopCh, c.crdSynced) { |
||||
return |
||||
} |
||||
|
||||
// process each item in the list once
|
||||
if crds, err := c.crdLister.List(labels.Everything()); err != nil { |
||||
utilruntime.HandleError(err) |
||||
} else { |
||||
for _, crd := range crds { |
||||
for _, version := range crd.Spec.Versions { |
||||
if err := c.syncHandler(schema.GroupVersion{Group: crd.Spec.Group, Version: version.Name}); err != nil { |
||||
utilruntime.HandleError(err) |
||||
} |
||||
} |
||||
} |
||||
} |
||||
close(c.syncedInitialSet) |
||||
|
||||
// start up your worker threads based on workers. Some controllers have multiple kinds of workers
|
||||
for i := 0; i < workers; i++ { |
||||
// runWorker will loop until "something bad" happens. The .Until will then rekick the worker
|
||||
// after one second
|
||||
go wait.Until(c.runWorker, time.Second, stopCh) |
||||
} |
||||
|
||||
// wait until we're told to stop
|
||||
<-stopCh |
||||
} |
||||
|
||||
// WaitForInitialSync blocks until the initial set of CRD resources has been processed
|
||||
func (c *crdRegistrationController) WaitForInitialSync() { |
||||
<-c.syncedInitialSet |
||||
} |
||||
|
||||
func (c *crdRegistrationController) runWorker() { |
||||
// hot loop until we're told to stop. processNextWorkItem will automatically wait until there's work
|
||||
// available, so we don't worry about secondary waits
|
||||
for c.processNextWorkItem() { |
||||
} |
||||
} |
||||
|
||||
// processNextWorkItem deals with one key off the queue. It returns false when it's time to quit.
|
||||
func (c *crdRegistrationController) processNextWorkItem() bool { |
||||
// pull the next work item from queue. It should be a key we use to lookup something in a cache
|
||||
key, quit := c.queue.Get() |
||||
if quit { |
||||
return false |
||||
} |
||||
// you always have to indicate to the queue that you've completed a piece of work
|
||||
defer c.queue.Done(key) |
||||
|
||||
// do your work on the key. This method will contains your "do stuff" logic
|
||||
err := c.syncHandler(key.(schema.GroupVersion)) |
||||
if err == nil { |
||||
// if you had no error, tell the queue to stop tracking history for your key. This will
|
||||
// reset things like failure counts for per-item rate limiting
|
||||
c.queue.Forget(key) |
||||
return true |
||||
} |
||||
|
||||
// there was a failure so be sure to report it. This method allows for pluggable error handling
|
||||
// which can be used for things like cluster-monitoring
|
||||
utilruntime.HandleError(fmt.Errorf("%v failed with : %v", key, err)) |
||||
// since we failed, we should requeue the item to work on later. This method will add a backoff
|
||||
// to avoid hotlooping on particular items (they're probably still not going to work right away)
|
||||
// and overall controller protection (everything I've done is broken, this controller needs to
|
||||
// calm down or it can starve other useful work) cases.
|
||||
c.queue.AddRateLimited(key) |
||||
|
||||
return true |
||||
} |
||||
|
||||
func (c *crdRegistrationController) enqueueCRD(crd *apiextensionsv1.CustomResourceDefinition) { |
||||
for _, version := range crd.Spec.Versions { |
||||
c.queue.Add(schema.GroupVersion{Group: crd.Spec.Group, Version: version.Name}) |
||||
} |
||||
} |
||||
|
||||
func (c *crdRegistrationController) handleVersionUpdate(groupVersion schema.GroupVersion) error { |
||||
apiServiceName := groupVersion.Version + "." + groupVersion.Group |
||||
|
||||
// check all CRDs. There shouldn't that many, but if we have problems later we can index them
|
||||
crds, err := c.crdLister.List(labels.Everything()) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
for _, crd := range crds { |
||||
if crd.Spec.Group != groupVersion.Group { |
||||
continue |
||||
} |
||||
for _, version := range crd.Spec.Versions { |
||||
if version.Name != groupVersion.Version || !version.Served { |
||||
continue |
||||
} |
||||
|
||||
c.apiServiceRegistration.AddAPIServiceToSync(&v1.APIService{ |
||||
ObjectMeta: metav1.ObjectMeta{Name: apiServiceName}, |
||||
Spec: v1.APIServiceSpec{ |
||||
Group: groupVersion.Group, |
||||
Version: groupVersion.Version, |
||||
GroupPriorityMinimum: 1000, // CRDs should have relatively low priority
|
||||
VersionPriority: 100, // CRDs will be sorted by kube-like versions like any other APIService with the same VersionPriority
|
||||
}, |
||||
}) |
||||
return nil |
||||
} |
||||
} |
||||
|
||||
c.apiServiceRegistration.RemoveAPIServiceToSync(apiServiceName) |
||||
return nil |
||||
} |
||||
@ -1,51 +0,0 @@ |
||||
# grafana aggregator |
||||
|
||||
The `aggregator` command in this binary is our equivalent of what kube-apiserver does for aggregation using |
||||
the `kube-aggregator` pkg. Here, we enable only select controllers that are useful for aggregation in a Grafana |
||||
cloud context. In future, Grafana microservices (and even plugins) will run as separate API servers |
||||
hosting each their own APIs (with specific Group/Versions). The `aggregator` component here shall act similar to what |
||||
`kube-apiserver` does: doing healthchecks for `APIService` objects registered against it and acting as a proxy for |
||||
the specified `GroupVersion` therein. |
||||
|
||||
## How to get started |
||||
|
||||
1. Generate the PKI using `openssl` (for development purposes, we will use the CN of `system:masters`): |
||||
```shell |
||||
./hack/make-aggregator-pki.sh |
||||
``` |
||||
2. Start the aggregator: |
||||
```shell |
||||
# This will generate the kubeconfig which you can use in the extension apiservers for |
||||
# enforcing delegate authnz under $PWD/data/grafana-apiserver/aggregator.kubeconfig |
||||
go run ./pkg/cmd/grafana aggregator --secure-port 8443 \ |
||||
--proxy-client-cert-file $PWD/data/grafana-aggregator/client.crt \ |
||||
--proxy-client-key-file $PWD/data/grafana-aggregator/client.key |
||||
``` |
||||
3. Apply the manifests: |
||||
```shell |
||||
export KUBECONFIG=$PWD/data/grafana-apiserver/aggregator.kubeconfig |
||||
kubectl apply -k ./pkg/cmd/grafana/apiserver/deploy/aggregator-test |
||||
# SAMPLE OUTPUT |
||||
# apiservice.apiregistration.k8s.io/v0alpha1.example.grafana.app created |
||||
# externalname.service.grafana.app/example-apiserver created |
||||
|
||||
kubectl get apiservice |
||||
# SAMPLE OUTPUT |
||||
# NAME SERVICE AVAILABLE AGE |
||||
# v0alpha1.example.grafana.app grafana/example-apiserver False (FailedDiscoveryCheck) 29m |
||||
``` |
||||
4. In another tab, start the example microservice that will be aggregated by the parent apiserver: |
||||
```shell |
||||
go run ./pkg/cmd/grafana apiserver example.grafana.app \ |
||||
--kubeconfig $PWD/data/grafana-aggregator/aggregator.kubeconfig \ |
||||
--secure-port 7443 \ |
||||
--client-ca-file=$PWD/data/grafana-aggregator/ca.crt |
||||
``` |
||||
5. Check `APIService` again: |
||||
```shell |
||||
export KUBECONFIG=$PWD/data/grafana-apiserver/aggregator.kubeconfig |
||||
kubectl get apiservice |
||||
# SAMPLE OUTPUT |
||||
# NAME SERVICE AVAILABLE AGE |
||||
# v0alpha1.example.grafana.app grafana/example-apiserver True 30m |
||||
``` |
||||
@ -0,0 +1,148 @@ |
||||
package datasource |
||||
|
||||
import ( |
||||
"context" |
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend" |
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" |
||||
|
||||
common "github.com/grafana/grafana/pkg/apis/common/v0alpha1" |
||||
"github.com/grafana/grafana/pkg/apis/datasource/v0alpha1" |
||||
"github.com/grafana/grafana/pkg/infra/appcontext" |
||||
"github.com/grafana/grafana/pkg/plugins" |
||||
"github.com/grafana/grafana/pkg/services/apiserver/endpoints/request" |
||||
"github.com/grafana/grafana/pkg/services/datasources" |
||||
) |
||||
|
||||
type QuerierFactoryFunc func(ctx context.Context, ri common.ResourceInfo, pj plugins.JSONData) (Querier, error) |
||||
|
||||
type QuerierProvider interface { |
||||
Querier(ctx context.Context, ri common.ResourceInfo, pj plugins.JSONData) (Querier, error) |
||||
} |
||||
|
||||
type DefaultQuerierProvider struct { |
||||
factory QuerierFactoryFunc |
||||
} |
||||
|
||||
func ProvideDefaultQuerierProvider(pluginClient plugins.Client, dsService datasources.DataSourceService, |
||||
dsCache datasources.CacheService) *DefaultQuerierProvider { |
||||
return NewQuerierProvider(func(ctx context.Context, ri common.ResourceInfo, pj plugins.JSONData) (Querier, error) { |
||||
return NewDefaultQuerier(ri, pj, pluginClient, dsService, dsCache), nil |
||||
}) |
||||
} |
||||
|
||||
func NewQuerierProvider(factory QuerierFactoryFunc) *DefaultQuerierProvider { |
||||
return &DefaultQuerierProvider{ |
||||
factory: factory, |
||||
} |
||||
} |
||||
|
||||
func (p *DefaultQuerierProvider) Querier(ctx context.Context, ri common.ResourceInfo, pj plugins.JSONData) (Querier, error) { |
||||
return p.factory(ctx, ri, pj) |
||||
} |
||||
|
||||
// Querier is the interface that wraps the Query method.
|
||||
type Querier interface { |
||||
// Query runs the query on behalf of the user in context.
|
||||
Query(ctx context.Context, query *backend.QueryDataRequest) (*backend.QueryDataResponse, error) |
||||
// Health checks the health of the plugin.
|
||||
Health(ctx context.Context, req *backend.CheckHealthRequest) (*backend.CheckHealthResult, error) |
||||
// Resource gets a resource plugin.
|
||||
Resource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error |
||||
// Datasource gets all data source plugins (with elevated permissions).
|
||||
Datasource(ctx context.Context, name string) (*v0alpha1.DataSourceConnection, error) |
||||
// Datasources lists all data sources (with elevated permissions).
|
||||
Datasources(ctx context.Context) (*v0alpha1.DataSourceConnectionList, error) |
||||
} |
||||
|
||||
type DefaultQuerier struct { |
||||
connectionResourceInfo common.ResourceInfo |
||||
pluginJSON plugins.JSONData |
||||
pluginClient plugins.Client |
||||
dsService datasources.DataSourceService |
||||
dsCache datasources.CacheService |
||||
} |
||||
|
||||
func NewDefaultQuerier( |
||||
connectionResourceInfo common.ResourceInfo, |
||||
pluginJSON plugins.JSONData, |
||||
pluginClient plugins.Client, |
||||
dsService datasources.DataSourceService, |
||||
dsCache datasources.CacheService, |
||||
) *DefaultQuerier { |
||||
return &DefaultQuerier{ |
||||
connectionResourceInfo: connectionResourceInfo, |
||||
pluginJSON: pluginJSON, |
||||
pluginClient: pluginClient, |
||||
dsService: dsService, |
||||
dsCache: dsCache, |
||||
} |
||||
} |
||||
|
||||
func (q *DefaultQuerier) Query(ctx context.Context, query *backend.QueryDataRequest) (*backend.QueryDataResponse, error) { |
||||
_, err := request.NamespaceInfoFrom(ctx, true) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
return q.pluginClient.QueryData(ctx, query) |
||||
} |
||||
|
||||
func (q *DefaultQuerier) Resource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error { |
||||
_, err := request.NamespaceInfoFrom(ctx, true) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
return q.pluginClient.CallResource(ctx, req, sender) |
||||
} |
||||
|
||||
func (q *DefaultQuerier) Health(ctx context.Context, query *backend.CheckHealthRequest) (*backend.CheckHealthResult, error) { |
||||
_, err := request.NamespaceInfoFrom(ctx, true) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
return q.pluginClient.CheckHealth(ctx, query) |
||||
} |
||||
|
||||
func (q *DefaultQuerier) Datasource(ctx context.Context, name string) (*v0alpha1.DataSourceConnection, error) { |
||||
info, err := request.NamespaceInfoFrom(ctx, true) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
user, err := appcontext.User(ctx) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
ds, err := q.dsCache.GetDatasourceByUID(ctx, name, user, false) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
return asConnection(ds, info.Value) |
||||
} |
||||
|
||||
func (q *DefaultQuerier) Datasources(ctx context.Context) (*v0alpha1.DataSourceConnectionList, error) { |
||||
info, err := request.NamespaceInfoFrom(ctx, true) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
ds, err := q.dsService.GetDataSourcesByType(ctx, &datasources.GetDataSourcesByTypeQuery{ |
||||
OrgID: info.OrgID, |
||||
Type: q.pluginJSON.ID, |
||||
}) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
return asConnectionList(q.connectionResourceInfo.TypeMeta(), ds, info.Value) |
||||
} |
||||
|
||||
func asConnectionList(typeMeta metav1.TypeMeta, dss []*datasources.DataSource, ns string) (*v0alpha1.DataSourceConnectionList, error) { |
||||
result := &v0alpha1.DataSourceConnectionList{ |
||||
Items: []v0alpha1.DataSourceConnection{}, |
||||
} |
||||
for _, ds := range dss { |
||||
v, _ := asConnection(ds, ns) |
||||
result.Items = append(result.Items, *v) |
||||
} |
||||
|
||||
return result, nil |
||||
} |
||||
@ -0,0 +1,285 @@ |
||||
// SPDX-License-Identifier: AGPL-3.0-only
|
||||
// Provenance-includes-location: https://github.com/kubernetes/kubernetes/blob/master/cmd/kube-apiserver/app/aggregator.go
|
||||
// Provenance-includes-license: Apache-2.0
|
||||
// Provenance-includes-copyright: The Kubernetes Authors.
|
||||
// Provenance-includes-location: https://github.com/kubernetes/kubernetes/blob/master/cmd/kube-apiserver/app/server.go
|
||||
// Provenance-includes-license: Apache-2.0
|
||||
// Provenance-includes-copyright: The Kubernetes Authors.
|
||||
// Provenance-includes-location: https://github.com/kubernetes/kubernetes/blob/master/pkg/controlplane/apiserver/apiextensions.go
|
||||
// Provenance-includes-license: Apache-2.0
|
||||
// Provenance-includes-copyright: The Kubernetes Authors.
|
||||
|
||||
package aggregator |
||||
|
||||
import ( |
||||
"crypto/tls" |
||||
"fmt" |
||||
"net/http" |
||||
"strings" |
||||
"sync" |
||||
"time" |
||||
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" |
||||
"k8s.io/apimachinery/pkg/runtime/schema" |
||||
utilnet "k8s.io/apimachinery/pkg/util/net" |
||||
"k8s.io/apimachinery/pkg/util/sets" |
||||
genericapiserver "k8s.io/apiserver/pkg/server" |
||||
"k8s.io/apiserver/pkg/server/healthz" |
||||
"k8s.io/client-go/informers" |
||||
"k8s.io/client-go/kubernetes/fake" |
||||
"k8s.io/client-go/tools/cache" |
||||
"k8s.io/klog/v2" |
||||
v1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1" |
||||
v1helper "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1/helper" |
||||
aggregatorapiserver "k8s.io/kube-aggregator/pkg/apiserver" |
||||
apiregistrationclientset "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset" |
||||
apiregistrationclient "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset/typed/apiregistration/v1" |
||||
apiregistrationInformers "k8s.io/kube-aggregator/pkg/client/informers/externalversions/apiregistration/v1" |
||||
"k8s.io/kube-aggregator/pkg/controllers/autoregister" |
||||
|
||||
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) { |
||||
// 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 |
||||
} |
||||
sharedInformerFactory := informersv0alpha1.NewSharedInformerFactory( |
||||
serviceClient, |
||||
5*time.Minute, // this is effectively used as a refresh interval right now. Might want to do something nicer later on.
|
||||
) |
||||
serviceResolver := NewExternalNameResolver(sharedInformerFactory.Service().V0alpha1().ExternalNames().Lister()) |
||||
|
||||
aggregatorConfig := &aggregatorapiserver.Config{ |
||||
GenericConfig: &genericapiserver.RecommendedConfig{ |
||||
Config: sharedConfig.Config, |
||||
SharedInformerFactory: fakev1Informers, |
||||
ClientConfig: sharedConfig.LoopbackClientConfig, |
||||
}, |
||||
ExtraConfig: aggregatorapiserver.ExtraConfig{ |
||||
ProxyClientCertFile: commandOptions.AggregatorOptions.ProxyClientCertFile, |
||||
ProxyClientKeyFile: commandOptions.AggregatorOptions.ProxyClientKeyFile, |
||||
// NOTE: while ProxyTransport can be skipped in the configuration, it allows honoring
|
||||
// DISABLE_HTTP2, HTTPS_PROXY and NO_PROXY env vars as needed
|
||||
ProxyTransport: createProxyTransport(), |
||||
ServiceResolver: serviceResolver, |
||||
}, |
||||
} |
||||
|
||||
if err := commandOptions.AggregatorOptions.ApplyTo(aggregatorConfig, commandOptions.RecommendedOptions.Etcd, commandOptions.StorageOptions.DataPath); err != nil { |
||||
return nil, nil, err |
||||
} |
||||
|
||||
return aggregatorConfig, sharedInformerFactory, nil |
||||
} |
||||
|
||||
func CreateAggregatorServer(aggregatorConfig *aggregatorapiserver.Config, sharedInformerFactory informersv0alpha1.SharedInformerFactory, delegateAPIServer genericapiserver.DelegationTarget) (*aggregatorapiserver.APIAggregator, error) { |
||||
completedConfig := aggregatorConfig.Complete() |
||||
aggregatorServer, err := completedConfig.NewWithDelegate(delegateAPIServer) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
// create controllers for auto-registration
|
||||
apiRegistrationClient, err := apiregistrationclient.NewForConfig(completedConfig.GenericConfig.LoopbackClientConfig) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
autoRegistrationController := autoregister.NewAutoRegisterController(aggregatorServer.APIRegistrationInformers.Apiregistration().V1().APIServices(), apiRegistrationClient) |
||||
apiServices := apiServicesToRegister(delegateAPIServer, autoRegistrationController) |
||||
|
||||
// Imbue all builtin group-priorities onto the aggregated discovery
|
||||
if completedConfig.GenericConfig.AggregatedDiscoveryGroupManager != nil { |
||||
for gv, entry := range APIVersionPriorities { |
||||
completedConfig.GenericConfig.AggregatedDiscoveryGroupManager.SetGroupVersionPriority(metav1.GroupVersion(gv), int(entry.Group), int(entry.Version)) |
||||
} |
||||
} |
||||
|
||||
err = aggregatorServer.GenericAPIServer.AddPostStartHook("grafana-apiserver-autoregistration", func(context genericapiserver.PostStartHookContext) error { |
||||
go func() { |
||||
autoRegistrationController.Run(5, context.StopCh) |
||||
}() |
||||
return nil |
||||
}) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
err = aggregatorServer.GenericAPIServer.AddBootSequenceHealthChecks( |
||||
makeAPIServiceAvailableHealthCheck( |
||||
"autoregister-completion", |
||||
apiServices, |
||||
aggregatorServer.APIRegistrationInformers.Apiregistration().V1().APIServices(), |
||||
), |
||||
) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
apiregistrationClient, err := apiregistrationclientset.NewForConfig(completedConfig.GenericConfig.LoopbackClientConfig) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
availableController, err := NewAvailableConditionController( |
||||
aggregatorServer.APIRegistrationInformers.Apiregistration().V1().APIServices(), |
||||
sharedInformerFactory.Service().V0alpha1().ExternalNames(), |
||||
apiregistrationClient.ApiregistrationV1(), |
||||
nil, |
||||
(func() ([]byte, []byte))(nil), |
||||
completedConfig.ExtraConfig.ServiceResolver, |
||||
) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
aggregatorServer.GenericAPIServer.AddPostStartHookOrDie("apiservice-status-override-available-controller", func(context genericapiserver.PostStartHookContext) error { |
||||
// if we end up blocking for long periods of time, we may need to increase workers.
|
||||
go availableController.Run(5, context.StopCh) |
||||
return nil |
||||
}) |
||||
|
||||
aggregatorServer.GenericAPIServer.AddPostStartHookOrDie("start-grafana-aggregator-informers", func(context genericapiserver.PostStartHookContext) error { |
||||
sharedInformerFactory.Start(context.StopCh) |
||||
aggregatorServer.APIRegistrationInformers.Start(context.StopCh) |
||||
return nil |
||||
}) |
||||
|
||||
return aggregatorServer, nil |
||||
} |
||||
|
||||
func makeAPIService(gv schema.GroupVersion) *v1.APIService { |
||||
apiServicePriority, ok := APIVersionPriorities[gv] |
||||
if !ok { |
||||
// if we aren't found, then we shouldn't register ourselves because it could result in a CRD group version
|
||||
// being permanently stuck in the APIServices list.
|
||||
klog.Infof("Skipping APIService creation for %v", gv) |
||||
return nil |
||||
} |
||||
return &v1.APIService{ |
||||
ObjectMeta: metav1.ObjectMeta{Name: gv.Version + "." + gv.Group}, |
||||
Spec: v1.APIServiceSpec{ |
||||
Group: gv.Group, |
||||
Version: gv.Version, |
||||
GroupPriorityMinimum: apiServicePriority.Group, |
||||
VersionPriority: apiServicePriority.Version, |
||||
}, |
||||
} |
||||
} |
||||
|
||||
// makeAPIServiceAvailableHealthCheck returns a healthz check that returns healthy
|
||||
// once all of the specified services have been observed to be available at least once.
|
||||
func makeAPIServiceAvailableHealthCheck(name string, apiServices []*v1.APIService, apiServiceInformer apiregistrationInformers.APIServiceInformer) healthz.HealthChecker { |
||||
// Track the auto-registered API services that have not been observed to be available yet
|
||||
pendingServiceNamesLock := &sync.RWMutex{} |
||||
pendingServiceNames := sets.NewString() |
||||
for _, service := range apiServices { |
||||
pendingServiceNames.Insert(service.Name) |
||||
} |
||||
|
||||
// When an APIService in the list is seen as available, remove it from the pending list
|
||||
handleAPIServiceChange := func(service *v1.APIService) { |
||||
pendingServiceNamesLock.Lock() |
||||
defer pendingServiceNamesLock.Unlock() |
||||
if !pendingServiceNames.Has(service.Name) { |
||||
return |
||||
} |
||||
if v1helper.IsAPIServiceConditionTrue(service, v1.Available) { |
||||
pendingServiceNames.Delete(service.Name) |
||||
} |
||||
} |
||||
|
||||
// Watch add/update events for APIServices
|
||||
_, _ = apiServiceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ |
||||
AddFunc: func(obj interface{}) { handleAPIServiceChange(obj.(*v1.APIService)) }, |
||||
UpdateFunc: func(old, new interface{}) { handleAPIServiceChange(new.(*v1.APIService)) }, |
||||
}) |
||||
|
||||
// Don't return healthy until the pending list is empty
|
||||
return healthz.NamedCheck(name, func(r *http.Request) error { |
||||
pendingServiceNamesLock.RLock() |
||||
defer pendingServiceNamesLock.RUnlock() |
||||
if pendingServiceNames.Len() > 0 { |
||||
return fmt.Errorf("missing APIService: %v", pendingServiceNames.List()) |
||||
} |
||||
return nil |
||||
}) |
||||
} |
||||
|
||||
// Priority defines group Priority that is used in discovery. This controls
|
||||
// group position in the kubectl output.
|
||||
type Priority struct { |
||||
// Group indicates the order of the Group relative to other groups.
|
||||
Group int32 |
||||
// Version indicates the relative order of the Version inside of its group.
|
||||
Version int32 |
||||
} |
||||
|
||||
// APIVersionPriorities are the proper way to resolve this letting the aggregator know the desired group and version-within-group order of the underlying servers
|
||||
// is to refactor the genericapiserver.DelegationTarget to include a list of priorities based on which APIs were installed.
|
||||
// This requires the APIGroupInfo struct to evolve and include the concept of priorities and to avoid mistakes, the core storage map there needs to be updated.
|
||||
// That ripples out every bit as far as you'd expect, so for 1.7 we'll include the list here instead of being built up during storage.
|
||||
var APIVersionPriorities = map[schema.GroupVersion]Priority{ |
||||
{Group: "", Version: "v1"}: {Group: 18000, Version: 1}, |
||||
// to my knowledge, nothing below here collides
|
||||
{Group: "admissionregistration.k8s.io", Version: "v1"}: {Group: 16700, Version: 15}, |
||||
{Group: "admissionregistration.k8s.io", Version: "v1beta1"}: {Group: 16700, Version: 12}, |
||||
{Group: "admissionregistration.k8s.io", Version: "v1alpha1"}: {Group: 16700, Version: 9}, |
||||
// Append a new group to the end of the list if unsure.
|
||||
// You can use min(existing group)-100 as the initial value for a group.
|
||||
// Version can be set to 9 (to have space around) for a new group.
|
||||
} |
||||
|
||||
func apiServicesToRegister(delegateAPIServer genericapiserver.DelegationTarget, registration autoregister.AutoAPIServiceRegistration) []*v1.APIService { |
||||
apiServices := []*v1.APIService{} |
||||
|
||||
for _, curr := range delegateAPIServer.ListedPaths() { |
||||
if curr == "/api/v1" { |
||||
apiService := makeAPIService(schema.GroupVersion{Group: "", Version: "v1"}) |
||||
registration.AddAPIServiceToSyncOnStart(apiService) |
||||
apiServices = append(apiServices, apiService) |
||||
continue |
||||
} |
||||
|
||||
if !strings.HasPrefix(curr, "/apis/") { |
||||
continue |
||||
} |
||||
// this comes back in a list that looks like /apis/rbac.authorization.k8s.io/v1alpha1
|
||||
tokens := strings.Split(curr, "/") |
||||
if len(tokens) != 4 { |
||||
continue |
||||
} |
||||
|
||||
apiService := makeAPIService(schema.GroupVersion{Group: tokens[2], Version: tokens[3]}) |
||||
if apiService == nil { |
||||
continue |
||||
} |
||||
registration.AddAPIServiceToSyncOnStart(apiService) |
||||
apiServices = append(apiServices, apiService) |
||||
} |
||||
|
||||
return apiServices |
||||
} |
||||
|
||||
// NOTE: below function imported from https://github.com/kubernetes/kubernetes/blob/master/cmd/kube-apiserver/app/server.go#L197
|
||||
// createProxyTransport creates the dialer infrastructure to connect to the api servers.
|
||||
func createProxyTransport() *http.Transport { |
||||
// NOTE: We don't set proxyDialerFn but the below SetTransportDefaults will
|
||||
// See https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apimachinery/pkg/util/net/http.go#L109
|
||||
var proxyDialerFn utilnet.DialFunc |
||||
// Proxying to services is IP-based... don't expect to be able to verify the hostname
|
||||
proxyTLSClientConfig := &tls.Config{InsecureSkipVerify: true} |
||||
proxyTransport := utilnet.SetTransportDefaults(&http.Transport{ |
||||
DialContext: proxyDialerFn, |
||||
TLSClientConfig: proxyTLSClientConfig, |
||||
}) |
||||
return proxyTransport |
||||
} |
||||
@ -0,0 +1,48 @@ |
||||
package apiserver |
||||
|
||||
import ( |
||||
"fmt" |
||||
"net" |
||||
"path/filepath" |
||||
"strconv" |
||||
|
||||
"github.com/grafana/grafana/pkg/services/apiserver/options" |
||||
"github.com/grafana/grafana/pkg/services/featuremgmt" |
||||
"github.com/grafana/grafana/pkg/setting" |
||||
) |
||||
|
||||
func applyGrafanaConfig(cfg *setting.Cfg, features featuremgmt.FeatureToggles, o *options.Options) { |
||||
defaultLogLevel := 0 |
||||
ip := net.ParseIP(cfg.HTTPAddr) |
||||
apiURL := cfg.AppURL |
||||
port, err := strconv.Atoi(cfg.HTTPPort) |
||||
if err != nil { |
||||
port = 3000 |
||||
} |
||||
|
||||
if cfg.Env == setting.Dev { |
||||
defaultLogLevel = 10 |
||||
port = 6443 |
||||
ip = net.ParseIP("127.0.0.1") |
||||
apiURL = fmt.Sprintf("https://%s:%d", ip, port) |
||||
} |
||||
|
||||
host := fmt.Sprintf("%s:%d", ip, port) |
||||
|
||||
o.RecommendedOptions.Etcd.StorageConfig.Transport.ServerList = cfg.SectionWithEnvOverrides("grafana-apiserver").Key("etcd_servers").Strings(",") |
||||
|
||||
o.RecommendedOptions.SecureServing.BindAddress = ip |
||||
o.RecommendedOptions.SecureServing.BindPort = port |
||||
o.RecommendedOptions.Authentication.RemoteKubeConfigFileOptional = true |
||||
o.RecommendedOptions.Authorization.RemoteKubeConfigFileOptional = true |
||||
|
||||
o.RecommendedOptions.Admission = nil |
||||
o.RecommendedOptions.CoreAPI = nil |
||||
|
||||
o.StorageOptions.StorageType = options.StorageType(cfg.SectionWithEnvOverrides("grafana-apiserver").Key("storage_type").MustString(string(options.StorageTypeLegacy))) |
||||
o.StorageOptions.DataPath = filepath.Join(cfg.DataPath, "grafana-apiserver") |
||||
o.ExtraOptions.DevMode = features.IsEnabledGlobally(featuremgmt.FlagGrafanaAPIServerEnsureKubectlAccess) |
||||
o.ExtraOptions.ExternalAddress = host |
||||
o.ExtraOptions.APIURL = apiURL |
||||
o.ExtraOptions.Verbosity = defaultLogLevel |
||||
} |
||||
@ -0,0 +1,112 @@ |
||||
package options |
||||
|
||||
import ( |
||||
"github.com/spf13/pflag" |
||||
v1 "k8s.io/api/apps/v1" |
||||
"k8s.io/apimachinery/pkg/runtime" |
||||
"k8s.io/apimachinery/pkg/runtime/schema" |
||||
openapinamer "k8s.io/apiserver/pkg/endpoints/openapi" |
||||
genericfeatures "k8s.io/apiserver/pkg/features" |
||||
genericapiserver "k8s.io/apiserver/pkg/server" |
||||
"k8s.io/apiserver/pkg/server/options" |
||||
"k8s.io/apiserver/pkg/server/resourceconfig" |
||||
utilfeature "k8s.io/apiserver/pkg/util/feature" |
||||
apiregistrationv1beta1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1beta1" |
||||
aggregatorapiserver "k8s.io/kube-aggregator/pkg/apiserver" |
||||
aggregatorscheme "k8s.io/kube-aggregator/pkg/apiserver/scheme" |
||||
aggregatoropenapi "k8s.io/kube-aggregator/pkg/generated/openapi" |
||||
"k8s.io/kube-openapi/pkg/common" |
||||
|
||||
servicev0alpha1 "github.com/grafana/grafana/pkg/apis/service/v0alpha1" |
||||
filestorage "github.com/grafana/grafana/pkg/services/apiserver/storage/file" |
||||
) |
||||
|
||||
// AggregatorServerOptions contains the state for the aggregator apiserver
|
||||
type AggregatorServerOptions struct { |
||||
AlternateDNS []string |
||||
ProxyClientCertFile string |
||||
ProxyClientKeyFile string |
||||
} |
||||
|
||||
func NewAggregatorServerOptions() *AggregatorServerOptions { |
||||
return &AggregatorServerOptions{} |
||||
} |
||||
|
||||
func (o *AggregatorServerOptions) getMergedOpenAPIDefinitions(ref common.ReferenceCallback) map[string]common.OpenAPIDefinition { |
||||
aggregatorAPIs := aggregatoropenapi.GetOpenAPIDefinitions(ref) |
||||
return aggregatorAPIs |
||||
} |
||||
|
||||
func (o *AggregatorServerOptions) AddFlags(fs *pflag.FlagSet) { |
||||
if o == nil { |
||||
return |
||||
} |
||||
|
||||
fs.StringVar(&o.ProxyClientCertFile, "proxy-client-cert-file", o.ProxyClientCertFile, |
||||
"path to proxy client cert file") |
||||
|
||||
fs.StringVar(&o.ProxyClientKeyFile, "proxy-client-key-file", o.ProxyClientKeyFile, |
||||
"path to proxy client cert file") |
||||
} |
||||
|
||||
func (o *AggregatorServerOptions) Validate() []error { |
||||
if o == nil { |
||||
return nil |
||||
} |
||||
|
||||
// TODO: do we need to validate anything here?
|
||||
return nil |
||||
} |
||||
|
||||
func (o *AggregatorServerOptions) ApplyTo(aggregatorConfig *aggregatorapiserver.Config, etcdOpts *options.EtcdOptions, dataPath string) error { |
||||
genericConfig := aggregatorConfig.GenericConfig |
||||
|
||||
genericConfig.PostStartHooks = map[string]genericapiserver.PostStartHookConfigEntry{} |
||||
genericConfig.RESTOptionsGetter = nil |
||||
|
||||
if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.StorageVersionAPI) && |
||||
utilfeature.DefaultFeatureGate.Enabled(genericfeatures.APIServerIdentity) { |
||||
// Add StorageVersionPrecondition handler to aggregator-apiserver.
|
||||
// The handler will block write requests to built-in resources until the
|
||||
// target resources' storage versions are up-to-date.
|
||||
genericConfig.BuildHandlerChainFunc = genericapiserver.BuildHandlerChainWithStorageVersionPrecondition |
||||
} |
||||
|
||||
// copy the etcd options so we don't mutate originals.
|
||||
// we assume that the etcd options have been completed already. avoid messing with anything outside
|
||||
// of changes to StorageConfig as that may lead to unexpected behavior when the options are applied.
|
||||
etcdOptions := *etcdOpts |
||||
etcdOptions.StorageConfig.Codec = aggregatorscheme.Codecs.LegacyCodec(v1.SchemeGroupVersion, |
||||
apiregistrationv1beta1.SchemeGroupVersion, |
||||
servicev0alpha1.SchemeGroupVersion) |
||||
etcdOptions.StorageConfig.EncodeVersioner = runtime.NewMultiGroupVersioner(v1.SchemeGroupVersion, |
||||
schema.GroupKind{Group: apiregistrationv1beta1.GroupName}, |
||||
schema.GroupKind{Group: servicev0alpha1.GROUP}) |
||||
etcdOptions.SkipHealthEndpoints = true // avoid double wiring of health checks
|
||||
if err := etcdOptions.ApplyTo(&genericConfig.Config); err != nil { |
||||
return err |
||||
} |
||||
// override the RESTOptionsGetter to use the file storage options getter
|
||||
aggregatorConfig.GenericConfig.RESTOptionsGetter = filestorage.NewRESTOptionsGetter(dataPath, etcdOptions.StorageConfig) |
||||
|
||||
// prevent generic API server from installing the OpenAPI handler. Aggregator server has its own customized OpenAPI handler.
|
||||
genericConfig.SkipOpenAPIInstallation = true |
||||
mergedResourceConfig, err := resourceconfig.MergeAPIResourceConfigs(aggregatorapiserver.DefaultAPIResourceConfigSource(), nil, aggregatorscheme.Scheme) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
genericConfig.MergedResourceConfig = mergedResourceConfig |
||||
|
||||
namer := openapinamer.NewDefinitionNamer(aggregatorscheme.Scheme) |
||||
genericConfig.OpenAPIV3Config = genericapiserver.DefaultOpenAPIV3Config(o.getMergedOpenAPIDefinitions, namer) |
||||
genericConfig.OpenAPIV3Config.Info.Title = "Kubernetes" |
||||
genericConfig.OpenAPIConfig = genericapiserver.DefaultOpenAPIConfig(o.getMergedOpenAPIDefinitions, namer) |
||||
genericConfig.OpenAPIConfig.Info.Title = "Kubernetes" |
||||
genericConfig.PostStartHooks = map[string]genericapiserver.PostStartHookConfigEntry{} |
||||
|
||||
// These hooks use v1 informers, which are not available in the grafana aggregator.
|
||||
genericConfig.DisabledPostStartHooks = genericConfig.DisabledPostStartHooks.Insert("apiservice-status-available-controller") |
||||
genericConfig.DisabledPostStartHooks = genericConfig.DisabledPostStartHooks.Insert("start-kube-aggregator-informers") |
||||
|
||||
return nil |
||||
} |
||||
@ -0,0 +1,46 @@ |
||||
package options |
||||
|
||||
import ( |
||||
"strconv" |
||||
|
||||
"github.com/go-logr/logr" |
||||
"github.com/spf13/pflag" |
||||
genericapiserver "k8s.io/apiserver/pkg/server" |
||||
"k8s.io/component-base/logs" |
||||
"k8s.io/klog/v2" |
||||
) |
||||
|
||||
type ExtraOptions struct { |
||||
DevMode bool |
||||
ExternalAddress string |
||||
APIURL string |
||||
Verbosity int |
||||
} |
||||
|
||||
func NewExtraOptions() *ExtraOptions { |
||||
return &ExtraOptions{ |
||||
DevMode: false, |
||||
Verbosity: 0, |
||||
} |
||||
} |
||||
|
||||
func (o *ExtraOptions) AddFlags(fs *pflag.FlagSet) { |
||||
fs.BoolVar(&o.DevMode, "grafana-apiserver-dev-mode", o.DevMode, "Enable dev mode") |
||||
fs.StringVar(&o.ExternalAddress, "grafana-apiserver-host", o.ExternalAddress, "Host") |
||||
fs.StringVar(&o.APIURL, "grafana-apiserver-api-url", o.APIURL, "API URL") |
||||
fs.IntVar(&o.Verbosity, "verbosity", o.Verbosity, "Verbosity") |
||||
} |
||||
|
||||
func (o *ExtraOptions) Validate() []error { |
||||
return nil |
||||
} |
||||
|
||||
func (o *ExtraOptions) ApplyTo(c *genericapiserver.RecommendedConfig) error { |
||||
logger := logr.New(newLogAdapter(o.Verbosity)) |
||||
klog.SetLoggerWithOptions(logger, klog.ContextualLogger(true)) |
||||
if _, err := logs.GlogSetter(strconv.Itoa(o.Verbosity)); err != nil { |
||||
logger.Error(err, "failed to set log level") |
||||
} |
||||
c.ExternalAddress = o.ExternalAddress |
||||
return nil |
||||
} |
||||
@ -1,4 +1,4 @@ |
||||
package grafanaapiserver |
||||
package options |
||||
|
||||
import ( |
||||
"strings" |
||||
@ -0,0 +1,126 @@ |
||||
package options |
||||
|
||||
import ( |
||||
"net" |
||||
|
||||
"github.com/spf13/pflag" |
||||
"k8s.io/apimachinery/pkg/runtime" |
||||
"k8s.io/apiserver/pkg/endpoints/discovery/aggregated" |
||||
genericapiserver "k8s.io/apiserver/pkg/server" |
||||
genericoptions "k8s.io/apiserver/pkg/server/options" |
||||
) |
||||
|
||||
const defaultEtcdPathPrefix = "/registry/grafana.app" |
||||
|
||||
type Options struct { |
||||
RecommendedOptions *genericoptions.RecommendedOptions |
||||
AggregatorOptions *AggregatorServerOptions |
||||
StorageOptions *StorageOptions |
||||
ExtraOptions *ExtraOptions |
||||
} |
||||
|
||||
func NewOptions(codec runtime.Codec) *Options { |
||||
return &Options{ |
||||
RecommendedOptions: genericoptions.NewRecommendedOptions( |
||||
defaultEtcdPathPrefix, |
||||
codec, |
||||
), |
||||
AggregatorOptions: NewAggregatorServerOptions(), |
||||
StorageOptions: NewStorageOptions(), |
||||
ExtraOptions: NewExtraOptions(), |
||||
} |
||||
} |
||||
|
||||
func (o *Options) AddFlags(fs *pflag.FlagSet) { |
||||
o.RecommendedOptions.AddFlags(fs) |
||||
o.AggregatorOptions.AddFlags(fs) |
||||
o.StorageOptions.AddFlags(fs) |
||||
o.ExtraOptions.AddFlags(fs) |
||||
} |
||||
|
||||
func (o *Options) Validate() []error { |
||||
if errs := o.ExtraOptions.Validate(); len(errs) != 0 { |
||||
return errs |
||||
} |
||||
|
||||
if errs := o.StorageOptions.Validate(); len(errs) != 0 { |
||||
return errs |
||||
} |
||||
|
||||
if errs := o.AggregatorOptions.Validate(); len(errs) != 0 { |
||||
return errs |
||||
} |
||||
|
||||
if errs := o.RecommendedOptions.SecureServing.Validate(); len(errs) != 0 { |
||||
return errs |
||||
} |
||||
|
||||
if errs := o.RecommendedOptions.Authentication.Validate(); len(errs) != 0 { |
||||
return errs |
||||
} |
||||
|
||||
if o.StorageOptions.StorageType == StorageTypeEtcd { |
||||
if errs := o.RecommendedOptions.Etcd.Validate(); len(errs) != 0 { |
||||
return errs |
||||
} |
||||
} |
||||
|
||||
return nil |
||||
} |
||||
|
||||
func (o *Options) ApplyTo(serverConfig *genericapiserver.RecommendedConfig) error { |
||||
serverConfig.AggregatedDiscoveryGroupManager = aggregated.NewResourceManager("apis") |
||||
|
||||
if err := o.ExtraOptions.ApplyTo(serverConfig); err != nil { |
||||
return err |
||||
} |
||||
|
||||
if !o.ExtraOptions.DevMode { |
||||
o.RecommendedOptions.SecureServing.Listener = newFakeListener() |
||||
} |
||||
|
||||
if err := o.RecommendedOptions.SecureServing.ApplyTo(&serverConfig.SecureServing, &serverConfig.LoopbackClientConfig); err != nil { |
||||
return err |
||||
} |
||||
|
||||
if err := o.RecommendedOptions.Authentication.ApplyTo(&serverConfig.Authentication, serverConfig.SecureServing, serverConfig.OpenAPIConfig); err != nil { |
||||
return err |
||||
} |
||||
|
||||
if !o.ExtraOptions.DevMode { |
||||
if err := serverConfig.SecureServing.Listener.Close(); err != nil { |
||||
return err |
||||
} |
||||
serverConfig.SecureServing = nil |
||||
} |
||||
|
||||
return nil |
||||
} |
||||
|
||||
type fakeListener struct { |
||||
server net.Conn |
||||
client net.Conn |
||||
} |
||||
|
||||
func newFakeListener() *fakeListener { |
||||
server, client := net.Pipe() |
||||
return &fakeListener{ |
||||
server: server, |
||||
client: client, |
||||
} |
||||
} |
||||
|
||||
func (f *fakeListener) Accept() (net.Conn, error) { |
||||
return f.server, nil |
||||
} |
||||
|
||||
func (f *fakeListener) Close() error { |
||||
if err := f.client.Close(); err != nil { |
||||
return err |
||||
} |
||||
return f.server.Close() |
||||
} |
||||
|
||||
func (f *fakeListener) Addr() net.Addr { |
||||
return &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 3000, Zone: ""} |
||||
} |
||||
@ -0,0 +1,51 @@ |
||||
package options |
||||
|
||||
import ( |
||||
"fmt" |
||||
|
||||
"github.com/spf13/pflag" |
||||
genericapiserver "k8s.io/apiserver/pkg/server" |
||||
"k8s.io/apiserver/pkg/server/options" |
||||
) |
||||
|
||||
type StorageType string |
||||
|
||||
const ( |
||||
StorageTypeFile StorageType = "file" |
||||
StorageTypeEtcd StorageType = "etcd" |
||||
StorageTypeLegacy StorageType = "legacy" |
||||
StorageTypeUnified StorageType = "unified" |
||||
StorageTypeUnifiedGrpc StorageType = "unified-grpc" |
||||
) |
||||
|
||||
type StorageOptions struct { |
||||
StorageType StorageType |
||||
DataPath string |
||||
} |
||||
|
||||
func NewStorageOptions() *StorageOptions { |
||||
return &StorageOptions{ |
||||
StorageType: StorageTypeLegacy, |
||||
} |
||||
} |
||||
|
||||
func (o *StorageOptions) AddFlags(fs *pflag.FlagSet) { |
||||
fs.StringVar((*string)(&o.StorageType), "grafana-apiserver-storage-type", string(o.StorageType), "Storage type") |
||||
fs.StringVar((*string)(&o.StorageType), "grafana-apiserver-storage-path", string(o.StorageType), "Storage path for file storage") |
||||
} |
||||
|
||||
func (o *StorageOptions) Validate() []error { |
||||
errs := []error{} |
||||
switch o.StorageType { |
||||
case StorageTypeFile, StorageTypeEtcd, StorageTypeLegacy, StorageTypeUnified, StorageTypeUnifiedGrpc: |
||||
// no-op
|
||||
default: |
||||
errs = append(errs, fmt.Errorf("--grafana-apiserver-storage-type must be one of %s, %s, %s, %s, %s", StorageTypeFile, StorageTypeEtcd, StorageTypeLegacy, StorageTypeUnified, StorageTypeUnifiedGrpc)) |
||||
} |
||||
return errs |
||||
} |
||||
|
||||
func (o *StorageOptions) ApplyTo(serverConfig *genericapiserver.RecommendedConfig, etcdOptions *options.EtcdOptions) error { |
||||
// TODO: move storage setup here
|
||||
return nil |
||||
} |
||||
@ -1,13 +1,15 @@ |
||||
package grafanaapiserver |
||||
package apiserver |
||||
|
||||
import ( |
||||
"github.com/google/wire" |
||||
|
||||
"github.com/grafana/grafana/pkg/services/apiserver/builder" |
||||
) |
||||
|
||||
var WireSet = wire.NewSet( |
||||
ProvideService, |
||||
wire.Bind(new(RestConfigProvider), new(*service)), |
||||
wire.Bind(new(Service), new(*service)), |
||||
wire.Bind(new(APIRegistrar), new(*service)), |
||||
wire.Bind(new(DirectRestConfigProvider), new(*service)), |
||||
wire.Bind(new(builder.APIRegistrar), new(*service)), |
||||
) |
||||
@ -1,60 +0,0 @@ |
||||
package grafanaapiserver |
||||
|
||||
import ( |
||||
"fmt" |
||||
"net" |
||||
"path/filepath" |
||||
"strconv" |
||||
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt" |
||||
"github.com/grafana/grafana/pkg/setting" |
||||
) |
||||
|
||||
type config struct { |
||||
enabled bool |
||||
devMode bool |
||||
|
||||
ip net.IP |
||||
port int |
||||
host string |
||||
apiURL string |
||||
|
||||
storageType StorageType |
||||
|
||||
etcdServers []string |
||||
dataPath string |
||||
|
||||
logLevel int |
||||
} |
||||
|
||||
func newConfig(cfg *setting.Cfg, features featuremgmt.FeatureToggles) *config { |
||||
defaultLogLevel := 0 |
||||
ip := net.ParseIP(cfg.HTTPAddr) |
||||
apiURL := cfg.AppURL |
||||
port, err := strconv.Atoi(cfg.HTTPPort) |
||||
if err != nil { |
||||
port = 3000 |
||||
} |
||||
|
||||
if cfg.Env == setting.Dev { |
||||
defaultLogLevel = 10 |
||||
port = 6443 |
||||
ip = net.ParseIP("127.0.0.1") |
||||
apiURL = fmt.Sprintf("https://%s:%d", ip, port) |
||||
} |
||||
|
||||
host := fmt.Sprintf("%s:%d", ip, port) |
||||
|
||||
return &config{ |
||||
enabled: true, |
||||
devMode: features.IsEnabledGlobally(featuremgmt.FlagGrafanaAPIServerEnsureKubectlAccess), |
||||
dataPath: filepath.Join(cfg.DataPath, "grafana-apiserver"), |
||||
ip: ip, |
||||
port: port, |
||||
host: host, |
||||
logLevel: cfg.SectionWithEnvOverrides("grafana-apiserver").Key("log_level").MustInt(defaultLogLevel), |
||||
etcdServers: cfg.SectionWithEnvOverrides("grafana-apiserver").Key("etcd_servers").Strings(","), |
||||
storageType: StorageType(cfg.SectionWithEnvOverrides("grafana-apiserver").Key("storage_type").MustString(string(StorageTypeLegacy))), |
||||
apiURL: apiURL, |
||||
} |
||||
} |
||||
@ -1,40 +0,0 @@ |
||||
package grafanaapiserver |
||||
|
||||
import ( |
||||
"net" |
||||
"testing" |
||||
|
||||
"github.com/stretchr/testify/require" |
||||
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt" |
||||
"github.com/grafana/grafana/pkg/setting" |
||||
) |
||||
|
||||
func TestNewConfig(t *testing.T) { |
||||
cfg := setting.NewCfg() |
||||
cfg.Env = setting.Prod |
||||
cfg.DataPath = "/tmp/grafana" |
||||
cfg.HTTPAddr = "10.0.0.1" |
||||
cfg.HTTPPort = "4000" |
||||
cfg.AppURL = "http://test:4000" |
||||
|
||||
section := cfg.Raw.Section("grafana-apiserver") |
||||
section.Key("log_level").SetValue("5") |
||||
section.Key("etcd_servers").SetValue("http://localhost:2379") |
||||
|
||||
actual := newConfig(cfg, featuremgmt.WithFeatures()) |
||||
|
||||
expected := &config{ |
||||
enabled: true, |
||||
devMode: false, |
||||
storageType: StorageTypeLegacy, |
||||
etcdServers: []string{"http://localhost:2379"}, |
||||
apiURL: "http://test:4000", |
||||
ip: net.ParseIP("10.0.0.1"), |
||||
port: 4000, |
||||
host: "10.0.0.1:4000", |
||||
dataPath: "/tmp/grafana/grafana-apiserver", |
||||
logLevel: 5, |
||||
} |
||||
require.Equal(t, expected, actual) |
||||
} |
||||
Loading…
Reference in new issue