[Prototyping] Webhook managed by controller (#98085)

This commit is contained in:
Roberto Jiménez Sánchez authored and GitHub committed 2024-12-18 07:34:29 +03:00
1 parent 46b9d5fc9e
commit a6ed3d6fc5
21 files changed
+597 -466

No files matched your search

+14 -9
View File
@@ -43,15 +43,6 @@ type GitHubRepositoryConfig struct {
// Token for accessing the repository.
// TODO: this should be part of secrets and a simple reference.
Token string `json:"token,omitempty"`
// WebhookURL is the URL to send webhooks events to.
// By default, the system will generate a URL for you but you can use this one to run
// grafana locally and test the webhooks.
// TODO: This should be generated by the controller and not user-provided.
WebhookURL string `json:"webhookURL,omitempty"`
// WebhookSecret is the secret used to validate incoming webhooks.
// TODO: this should be generated by the controller and not user-provided.
WebhookSecret string `json:"webhookSecret,omitempty"`
// TODO: Do we want an SSH url instead maybe?
// TODO: On-prem GitHub Enterprise support?
@@ -131,11 +122,18 @@ type EditingOptions struct {
// This is expected never to be created by a kubectl call or similar, and is expected to rarely (if ever) be edited manually.
// As such, it is also a little less well structured than the spec, such as conditional-but-ever-present fields.
type RepositoryStatus struct {
// FIXME: this is temporary field until we use different methods in the controller for create and update
// Initialized is true when the repository has been initialized
Initialized bool `json:"initialized"`
// This will get updated with the current health status (and updated periodically)
Health HealthStatus `json:"health"`
// Sync information
Sync SyncStatus `json:"sync"`
// Webhook Information (if applicable)
Webhook *WebhookStatus `json:"webhook"`
}
type HealthStatus struct {
@@ -175,6 +173,13 @@ type SyncStatus struct {
Hash string `json:"hash,omitempty"`
}
type WebhookStatus struct {
ID int64 `json:"id,omitempty"`
URL string `json:"url,omitempty"`
Secret string `json:"secret,omitempty"`
SubscribedEvents []string `json:"subscribedEvents,omitempty"`
}
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
type RepositoryList struct {
metav1.TypeMeta `json:",inline"`
@@ -434,6 +434,11 @@ func (in *RepositoryStatus) DeepCopyInto(out *RepositoryStatus) {
*out = *in
in.Health.DeepCopyInto(&out.Health)
in.Sync.DeepCopyInto(&out.Sync)
if in.Webhook != nil {
in, out := &in.Webhook, &out.Webhook
*out = new(WebhookStatus)
(*in).DeepCopyInto(*out)
}
return
}
@@ -623,3 +628,24 @@ func (in *WebhookResponse) DeepCopyObject() runtime.Object {
}
return nil
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *WebhookStatus) DeepCopyInto(out *WebhookStatus) {
*out = *in
if in.SubscribedEvents != nil {
in, out := &in.SubscribedEvents, &out.SubscribedEvents
*out = make([]string, len(*in))
copy(*out, *in)
}
return
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WebhookStatus.
func (in *WebhookStatus) DeepCopy() *WebhookStatus {
if in == nil {
return nil
}
out := new(WebhookStatus)
in.DeepCopyInto(out)
return out
}
@@ -40,6 +40,7 @@ func GetOpenAPIDefinitions(ref common.ReferenceCallback) map[string]common.OpenA
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.SyncStatus": schema_pkg_apis_provisioning_v0alpha1_SyncStatus(ref),
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.TestResults": schema_pkg_apis_provisioning_v0alpha1_TestResults(ref),
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.WebhookResponse": schema_pkg_apis_provisioning_v0alpha1_WebhookResponse(ref),
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.WebhookStatus": schema_pkg_apis_provisioning_v0alpha1_WebhookStatus(ref),
}
}
@@ -240,20 +241,6 @@ func schema_pkg_apis_provisioning_v0alpha1_GitHubRepositoryConfig(ref common.Ref
Format: "",
},
},
"webhookURL": {
SchemaProps: spec.SchemaProps{
Description: "WebhookURL is the URL to send webhooks events to. By default, the system will generate a URL for you but you can use this one to run grafana locally and test the webhooks.",
Type: []string{"string"},
Format: "",
},
},
"webhookSecret": {
SchemaProps: spec.SchemaProps{
Description: "WebhookSecret is the secret used to validate incoming webhooks.",
Type: []string{"string"},
Format: "",
},
},
"branchWorkflow": {
SchemaProps: spec.SchemaProps{
Description: "Whether we should commit to change branches and use a Pull Request flow to achieve this. By default, this is false (i.e. we will commit straight to the main branch).",
@@ -897,6 +884,14 @@ func schema_pkg_apis_provisioning_v0alpha1_RepositoryStatus(ref common.Reference
Description: "The status of a Repository. This is expected never to be created by a kubectl call or similar, and is expected to rarely (if ever) be edited manually. As such, it is also a little less well structured than the spec, such as conditional-but-ever-present fields.",
Type: []string{"object"},
Properties: map[string]spec.Schema{
"initialized": {
SchemaProps: spec.SchemaProps{
Description: "FIXME: this is temporary field until we use different methods in the controller for create and update Initialized is true when the repository has been initialized",
Default: false,
Type: []string{"boolean"},
Format: "",
},
},
"health": {
SchemaProps: spec.SchemaProps{
Description: "This will get updated with the current health status (and updated periodically)",
@@ -911,12 +906,18 @@ func schema_pkg_apis_provisioning_v0alpha1_RepositoryStatus(ref common.Reference
Ref: ref("github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.SyncStatus"),
},
},
"webhook": {
SchemaProps: spec.SchemaProps{
Description: "Webhook Information (if applicable)",
Ref: ref("github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.WebhookStatus"),
},
},
},
Required: []string{"health", "sync"},
Required: []string{"initialized", "health", "sync", "webhook"},
},
},
Dependencies: []string{
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.HealthStatus", "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.SyncStatus"},
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.HealthStatus", "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.SyncStatus", "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.WebhookStatus"},
}
}
@@ -1315,3 +1316,47 @@ func schema_pkg_apis_provisioning_v0alpha1_WebhookResponse(ref common.ReferenceC
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.JobSpec"},
}
}
func schema_pkg_apis_provisioning_v0alpha1_WebhookStatus(ref common.ReferenceCallback) common.OpenAPIDefinition {
return common.OpenAPIDefinition{
Schema: spec.Schema{
SchemaProps: spec.SchemaProps{
Type: []string{"object"},
Properties: map[string]spec.Schema{
"id": {
SchemaProps: spec.SchemaProps{
Type: []string{"integer"},
Format: "int64",
},
},
"url": {
SchemaProps: spec.SchemaProps{
Type: []string{"string"},
Format: "",
},
},
"secret": {
SchemaProps: spec.SchemaProps{
Type: []string{"string"},
Format: "",
},
},
"subscribedEvents": {
SchemaProps: spec.SchemaProps{
Type: []string{"array"},
Items: &spec.SchemaOrArray{
Schema: &spec.Schema{
SchemaProps: spec.SchemaProps{
Default: "",
Type: []string{"string"},
Format: "",
},
},
},
},
},
},
},
},
}
}
@@ -5,6 +5,7 @@ API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provis
API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,ResourceWrapper,Lint
API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,SyncStatus,Message
API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,TestResults,Errors
API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,WebhookStatus,SubscribedEvents
API rule violation: names_match,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,FileList,Items
API rule violation: names_match,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,RepositorySpec,GitHub
API rule violation: names_match,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,RepositorySpec,PreferYAML
@@ -11,8 +11,6 @@ type GitHubRepositoryConfigApplyConfiguration struct {
Repository *string `json:"repository,omitempty"`
Branch *string `json:"branch,omitempty"`
Token *string `json:"token,omitempty"`
WebhookURL *string `json:"webhookURL,omitempty"`
WebhookSecret *string `json:"webhookSecret,omitempty"`
BranchWorkflow *bool `json:"branchWorkflow,omitempty"`
GenerateDashboardPreviews *bool `json:"generateDashboardPreviews,omitempty"`
PullRequestLinter *bool `json:"pullRequestLinter,omitempty"`
@@ -56,22 +54,6 @@ func (b *GitHubRepositoryConfigApplyConfiguration) WithToken(value string) *GitH
return b
}
// WithWebhookURL sets the WebhookURL field in the declarative configuration to the given value
// and returns the receiver, so that objects can be built by chaining "With" function invocations.
// If called multiple times, the WebhookURL field is set to the value of the last call.
func (b *GitHubRepositoryConfigApplyConfiguration) WithWebhookURL(value string) *GitHubRepositoryConfigApplyConfiguration {
b.WebhookURL = &value
return b
}
// WithWebhookSecret sets the WebhookSecret field in the declarative configuration to the given value
// and returns the receiver, so that objects can be built by chaining "With" function invocations.
// If called multiple times, the WebhookSecret field is set to the value of the last call.
func (b *GitHubRepositoryConfigApplyConfiguration) WithWebhookSecret(value string) *GitHubRepositoryConfigApplyConfiguration {
b.WebhookSecret = &value
return b
}
// WithBranchWorkflow sets the BranchWorkflow field in the declarative configuration to the given value
// and returns the receiver, so that objects can be built by chaining "With" function invocations.
// If called multiple times, the BranchWorkflow field is set to the value of the last call.
@@ -7,8 +7,10 @@ package v0alpha1
// RepositoryStatusApplyConfiguration represents a declarative configuration of the RepositoryStatus type for use
// with apply.
type RepositoryStatusApplyConfiguration struct {
Health *HealthStatusApplyConfiguration `json:"health,omitempty"`
Sync *SyncStatusApplyConfiguration `json:"sync,omitempty"`
Initialized *bool `json:"initialized,omitempty"`
Health *HealthStatusApplyConfiguration `json:"health,omitempty"`
Sync *SyncStatusApplyConfiguration `json:"sync,omitempty"`
Webhook *WebhookStatusApplyConfiguration `json:"webhook,omitempty"`
}
// RepositoryStatusApplyConfiguration constructs a declarative configuration of the RepositoryStatus type for use with
@@ -17,6 +19,14 @@ func RepositoryStatus() *RepositoryStatusApplyConfiguration {
return &RepositoryStatusApplyConfiguration{}
}
// WithInitialized sets the Initialized field in the declarative configuration to the given value
// and returns the receiver, so that objects can be built by chaining "With" function invocations.
// If called multiple times, the Initialized field is set to the value of the last call.
func (b *RepositoryStatusApplyConfiguration) WithInitialized(value bool) *RepositoryStatusApplyConfiguration {
b.Initialized = &value
return b
}
// WithHealth sets the Health field in the declarative configuration to the given value
// and returns the receiver, so that objects can be built by chaining "With" function invocations.
// If called multiple times, the Health field is set to the value of the last call.
@@ -32,3 +42,11 @@ func (b *RepositoryStatusApplyConfiguration) WithSync(value *SyncStatusApplyConf
b.Sync = value
return b
}
// WithWebhook sets the Webhook field in the declarative configuration to the given value
// and returns the receiver, so that objects can be built by chaining "With" function invocations.
// If called multiple times, the Webhook field is set to the value of the last call.
func (b *RepositoryStatusApplyConfiguration) WithWebhook(value *WebhookStatusApplyConfiguration) *RepositoryStatusApplyConfiguration {
b.Webhook = value
return b
}
@@ -0,0 +1,54 @@
// SPDX-License-Identifier: AGPL-3.0-only
// Code generated by applyconfiguration-gen. DO NOT EDIT.
package v0alpha1
// WebhookStatusApplyConfiguration represents a declarative configuration of the WebhookStatus type for use
// with apply.
type WebhookStatusApplyConfiguration struct {
ID *int64 `json:"id,omitempty"`
URL *string `json:"url,omitempty"`
Secret *string `json:"secret,omitempty"`
SubscribedEvents []string `json:"subscribedEvents,omitempty"`
}
// WebhookStatusApplyConfiguration constructs a declarative configuration of the WebhookStatus type for use with
// apply.
func WebhookStatus() *WebhookStatusApplyConfiguration {
return &WebhookStatusApplyConfiguration{}
}
// WithID sets the ID field in the declarative configuration to the given value
// and returns the receiver, so that objects can be built by chaining "With" function invocations.
// If called multiple times, the ID field is set to the value of the last call.
func (b *WebhookStatusApplyConfiguration) WithID(value int64) *WebhookStatusApplyConfiguration {
b.ID = &value
return b
}
// WithURL sets the URL field in the declarative configuration to the given value
// and returns the receiver, so that objects can be built by chaining "With" function invocations.
// If called multiple times, the URL field is set to the value of the last call.
func (b *WebhookStatusApplyConfiguration) WithURL(value string) *WebhookStatusApplyConfiguration {
b.URL = &value
return b
}
// WithSecret sets the Secret field in the declarative configuration to the given value
// and returns the receiver, so that objects can be built by chaining "With" function invocations.
// If called multiple times, the Secret field is set to the value of the last call.
func (b *WebhookStatusApplyConfiguration) WithSecret(value string) *WebhookStatusApplyConfiguration {
b.Secret = &value
return b
}
// WithSubscribedEvents adds the given value to the SubscribedEvents field in the declarative configuration
// and returns the receiver, so that objects can be build by chaining "With" function invocations.
// If called multiple times, values provided by each call will be appended to the SubscribedEvents field.
func (b *WebhookStatusApplyConfiguration) WithSubscribedEvents(values ...string) *WebhookStatusApplyConfiguration {
for i := range values {
b.SubscribedEvents = append(b.SubscribedEvents, values[i])
}
return b
}
@@ -38,6 +38,8 @@ func ForKind(kind schema.GroupVersionKind) interface{} {
return &provisioningv0alpha1.S3RepositoryConfigApplyConfiguration{}
case v0alpha1.SchemeGroupVersion.WithKind("SyncStatus"):
return &provisioningv0alpha1.SyncStatusApplyConfiguration{}
case v0alpha1.SchemeGroupVersion.WithKind("WebhookStatus"):
return &provisioningv0alpha1.WebhookStatusApplyConfiguration{}
// Group=service.grafana.app, Version=v0alpha1
case servicev0alpha1.SchemeGroupVersion.WithKind("ExternalName"):
+15 -182
View File
@@ -2,13 +2,9 @@ package provisioning
import (
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"fmt"
"log/slog"
"net/http"
"strings"
"time"
"github.com/prometheus/client_golang/prometheus"
@@ -20,7 +16,6 @@ import (
"k8s.io/apimachinery/pkg/util/validation/field"
"k8s.io/apiserver/pkg/admission"
"k8s.io/apiserver/pkg/authorization/authorizer"
"k8s.io/apiserver/pkg/registry/generic/registry"
"k8s.io/apiserver/pkg/registry/rest"
genericapiserver "k8s.io/apiserver/pkg/server"
"k8s.io/kube-openapi/pkg/common"
@@ -38,6 +33,7 @@ import (
"github.com/grafana/grafana/pkg/registry/apis/provisioning/repository/github"
"github.com/grafana/grafana/pkg/registry/apis/provisioning/resources"
"github.com/grafana/grafana/pkg/registry/apis/provisioning/safepath"
"github.com/grafana/grafana/pkg/registry/apis/provisioning/secrets"
"github.com/grafana/grafana/pkg/services/apiserver"
"github.com/grafana/grafana/pkg/services/apiserver/builder"
"github.com/grafana/grafana/pkg/services/featuremgmt"
@@ -46,10 +42,7 @@ import (
"github.com/grafana/grafana/pkg/storage/unified/blob"
)
const (
resourceNamePlaceholder = "$RESOURCE_NAME_PLACEHOLDER"
repoControllerWorkers = 1
)
const repoControllerWorkers = 1
var (
_ builder.APIGroupBuilder = (*ProvisioningAPIBuilder)(nil)
@@ -194,11 +187,6 @@ func (b *ProvisioningAPIBuilder) UpdateAPIGroupInfo(apiGroupInfo *genericapiserv
b.urlProvider,
))
repositoryStorage.AfterCreate = b.afterCreate
// AfterUpdate doesn't have the old object, so we have to use BeginUpdate
repositoryStorage.BeginUpdate = b.beginUpdate
repositoryStorage.AfterDelete = b.afterDelete
repositoryStatusStorage := grafanaregistry.NewRegistryStatusStore(opts.Scheme, repositoryStorage)
b.getter = repositoryStorage
@@ -311,13 +299,18 @@ func (b *ProvisioningAPIBuilder) AsRepository(ctx context.Context, r *provisioni
case provisioning.LocalRepositoryType:
return repository.NewLocal(r, b.localFileResolver), nil
case provisioning.GitHubRepositoryType:
// HACK: replace resource name placeholder in the webhook URL with the actual resource name
// In case the name was automatically generated by k8s
if r.Spec.GitHub.WebhookURL != "" {
r.Spec.GitHub.WebhookURL = strings.ReplaceAll(r.Spec.GitHub.WebhookURL, resourceNamePlaceholder, r.GetName())
}
return repository.NewGitHub(ctx, r, b.ghFactory), nil
gvr := provisioning.RepositoryResourceInfo.GroupVersionResource()
webhookURL := fmt.Sprintf(
"%sapis/%s/%s/namespaces/%s/%s/%s/webhook",
b.urlProvider(r.GetNamespace()),
gvr.Group,
gvr.Version,
r.GetNamespace(),
gvr.Resource,
r.GetName(),
)
secretsSvc := secrets.NewService(b.webhookSecretKey)
return repository.NewGitHub(ctx, r, b.ghFactory, secretsSvc, webhookURL), nil
case provisioning.S3RepositoryType:
return repository.NewS3(r), nil
default:
@@ -325,135 +318,6 @@ func (b *ProvisioningAPIBuilder) AsRepository(ctx context.Context, r *provisioni
}
}
func (b *ProvisioningAPIBuilder) afterCreate(obj runtime.Object, opts *metav1.CreateOptions) {
cfg, ok := obj.(*provisioning.Repository)
if !ok {
b.logger.Error("object is not *provisioning.Repository", "kind", obj.GetObjectKind().GroupVersionKind())
return
}
logger := b.logger.With("repository", cfg.GetName(), "namespace", cfg.GetNamespace())
ctx := context.Background()
repo, err := b.asRepository(ctx, cfg)
if err != nil {
logger.Error("failed to get repository", "error", err)
return
}
if err := b.ensureRepositoryFolderExists(ctx, cfg); err != nil {
logger.Error("failed to ensure repository folder exists", "error", err)
return
}
if err := repo.AfterCreate(ctx, b.logger); err != nil {
logger.Error("failed to run after create", "error", err)
return
}
}
func (b *ProvisioningAPIBuilder) beginUpdate(ctx context.Context, obj, old runtime.Object, opts *metav1.UpdateOptions) (registry.FinishFunc, error) {
objCfg, ok := obj.(*provisioning.Repository)
if !ok {
return nil, fmt.Errorf("new object is not *provisioning.Repository")
}
oldCfg, ok := old.(*provisioning.Repository)
if !ok {
return nil, fmt.Errorf("old object is not *provisioning.Repository")
}
repo, err := b.asRepository(ctx, objCfg)
if err != nil {
return nil, fmt.Errorf("get new repository: %w", err)
}
oldRepo, err := b.asRepository(ctx, oldCfg)
if err != nil {
return nil, fmt.Errorf("get old repository: %w", err)
}
if err := b.ensureRepositoryFolderExists(ctx, objCfg); err != nil {
return nil, fmt.Errorf("failed to ensure the configured folder '%s' in '%s' exists: %w", objCfg.GetName(), objCfg.GetNamespace(), err)
}
undo, err := repo.BeginUpdate(ctx, b.logger, oldRepo)
if err != nil {
b.logger.Warn("error in begin update", "err", err)
}
return func(ctx context.Context, success bool) {
if !success && undo != nil {
if err := undo(ctx); err != nil {
b.logger.ErrorContext(ctx, "failed to undo failed update", "error", err)
}
}
}, nil
}
func (b *ProvisioningAPIBuilder) ensureRepositoryFolderExists(ctx context.Context, cfg *provisioning.Repository) error {
if cfg.Spec.Folder == "" {
// The root folder can't not exist, so we don't have to do anything.
return nil
}
client, _, err := b.client.New(cfg.GetNamespace())
if err != nil {
return err
}
// FIXME: make sure folders are actually enabled in the apiserver.
folderIface := client.Resource(schema.GroupVersionResource{
Group: "folder.grafana.app",
Version: "v0alpha1",
Resource: "folders",
})
_, err = folderIface.Get(ctx, cfg.Spec.Folder, metav1.GetOptions{})
if err == nil {
// The folder exists and doesn't need to be created.
return nil
} else if !apierrors.IsNotFound(err) {
return fmt.Errorf("failed to search for existing repo folder: %w", err)
}
title := cfg.Spec.Title
if title == "" {
title = cfg.Spec.Folder
}
_, err = folderIface.Create(ctx, &unstructured.Unstructured{
Object: map[string]any{
"metadata": map[string]any{
"name": cfg.Spec.Folder,
"namespace": cfg.GetNamespace(),
},
"spec": map[string]any{
"title": title,
"description": "Repository-managed folder.",
},
},
}, metav1.CreateOptions{})
return err
}
func (b *ProvisioningAPIBuilder) afterDelete(obj runtime.Object, opts *metav1.DeleteOptions) {
ctx := context.Background()
cfg, ok := obj.(*provisioning.Repository)
if !ok {
b.logger.Error("object is not *provisioning.Repository")
return
}
repo, err := b.asRepository(ctx, cfg)
if err != nil {
b.logger.Error("failed to get repository", "error", err)
return
}
if err := repo.AfterDelete(ctx, b.logger); err != nil {
b.logger.Error("failed to run after delete", "error", err)
return
}
}
func (b *ProvisioningAPIBuilder) Mutate(ctx context.Context, a admission.Attributes, o admission.ObjectInterfaces) error {
obj := a.GetObject()
@@ -474,25 +338,6 @@ func (b *ProvisioningAPIBuilder) Mutate(ctx context.Context, a admission.Attribu
if r.Spec.GitHub.Branch == "" {
r.Spec.GitHub.Branch = "main"
}
if r.Spec.GitHub.WebhookURL == "" {
name := a.GetName()
// HACK: if the name is empty, it's probably auto-generated by k8s
// we insert a placeholder so we can replace it later
if name == "" {
name = resourceNamePlaceholder
}
gvr := provisioning.RepositoryResourceInfo.GroupVersionResource()
r.Spec.GitHub.WebhookURL = fmt.Sprintf("%sapis/%s/%s/namespaces/%s/%s/%s/webhook",
b.urlProvider(a.GetNamespace()), // gets the full name
gvr.Group, gvr.Version,
a.GetNamespace(), gvr.Resource, name)
}
if r.Spec.GitHub.WebhookSecret == "" {
r.Spec.GitHub.WebhookSecret = b.generateWebhookSecret(r.Spec.GitHub.Token)
}
}
return nil
@@ -564,6 +409,7 @@ func (b *ProvisioningAPIBuilder) GetPostStartHooks() (map[string]genericapiserve
b, // repoGetter
b.identities,
b.tester,
b.jobs,
)
if err != nil {
return err
@@ -785,16 +631,3 @@ spec:
return oas, nil
}
// generateWebhookSecret generates a webhook secret from a token
// using the configured secret key. The generated secret is consistent.
// TODO: this must be replaced by a app platform secrets once we have that.
func (b *ProvisioningAPIBuilder) generateWebhookSecret(token string) string {
secretKey := []byte(b.webhookSecretKey)
h := hmac.New(sha256.New, secretKey)
h.Write([]byte(token))
hashed := h.Sum(nil)
return hex.EncodeToString(hashed)
}
@@ -6,8 +6,8 @@ import (
"fmt"
"log/slog"
"net/http"
"net/url"
"regexp"
"slices"
"strings"
"github.com/google/go-github/v66/github"
@@ -19,10 +19,18 @@ import (
pgh "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository/github"
)
var subscribedEvents = []string{"push", "pull_request"}
type SecretsService interface {
Encrypt(ctx context.Context, data string) (string, error)
}
type githubRepository struct {
logger *slog.Logger
config *provisioning.Repository
gh pgh.Client
logger *slog.Logger
config *provisioning.Repository
gh pgh.Client
secrets SecretsService
webhookURL string
}
var _ Repository = (*githubRepository)(nil)
@@ -31,11 +39,15 @@ func NewGitHub(
ctx context.Context,
config *provisioning.Repository,
factory pgh.ClientFactory,
secrets SecretsService,
webhookURL string,
) *githubRepository {
return &githubRepository{
config: config,
logger: slog.Default().With("logger", "github-repository"),
gh: factory.New(ctx, config.Spec.GitHub.Token),
config: config,
logger: slog.Default().With("logger", "github-repository"),
gh: factory.New(ctx, config.Spec.GitHub.Token),
secrets: secrets,
webhookURL: webhookURL,
}
}
@@ -68,17 +80,6 @@ func (r *githubRepository) Validate() (list field.ErrorList) {
if gh.GenerateDashboardPreviews && !gh.BranchWorkflow {
list = append(list, field.Forbidden(field.NewPath("spec", "github", "token"), "to generate dashboard previews, you must activate the branch workflow"))
}
if gh.WebhookURL == "" {
list = append(list, field.Required(field.NewPath("spec", "github", "webhookURL"), "a webhook URL is required"))
}
if gh.WebhookSecret == "" {
list = append(list, field.Required(field.NewPath("spec", "github", "webhookSecret"), "a webhook secret is required"))
}
_, err := url.Parse(gh.WebhookURL)
if err != nil {
list = append(list, field.Invalid(field.NewPath("spec", "github", "webhookURL"), gh.WebhookURL, "invalid URL"))
}
return list
}
@@ -412,7 +413,11 @@ func (r *githubRepository) ensureBranchExists(ctx context.Context, branchName st
// Webhook implements Repository.
func (r *githubRepository) Webhook(ctx context.Context, logger *slog.Logger, req *http.Request) (*provisioning.WebhookResponse, error) {
payload, err := github.ValidatePayload(req, []byte(r.config.Spec.GitHub.WebhookSecret))
if r.config.Status.Webhook == nil {
return nil, fmt.Errorf("unexpected webhook request")
}
payload, err := github.ValidatePayload(req, []byte(r.config.Status.Webhook.Secret))
if err != nil {
return nil, apierrors.NewUnauthorized("invalid signature")
}
@@ -519,7 +524,7 @@ func (r *githubRepository) parsePullRequestEvent(event *github.PullRequestEvent)
}, nil
}
func (r *githubRepository) LatestRef(ctx context.Context) (string, error) {
func (r *githubRepository) LatestRef(ctx context.Context, logger *slog.Logger) (string, error) {
branch, err := r.gh.GetBranch(ctx, r.config.Spec.GitHub.Owner, r.config.Spec.GitHub.Repository, r.Config().Spec.GitHub.Branch)
if err != nil {
return "", fmt.Errorf("get branch: %w", err)
@@ -531,7 +536,7 @@ func (r *githubRepository) LatestRef(ctx context.Context) (string, error) {
func (r *githubRepository) CompareFiles(ctx context.Context, logger *slog.Logger, base, ref string) ([]FileChange, error) {
if ref == "" {
var err error
ref, err = r.LatestRef(ctx)
ref, err = r.LatestRef(ctx, logger)
if err != nil {
return nil, fmt.Errorf("get latest ref: %w", err)
}
@@ -612,130 +617,147 @@ func (r *githubRepository) CommentPullRequestFile(ctx context.Context, logger *s
return r.gh.CreatePullRequestFileComment(ctx, r.config.Spec.GitHub.Owner, r.config.Spec.GitHub.Repository, prNumber, fileComment)
}
func (r *githubRepository) createWebhook(ctx context.Context, logger *slog.Logger) error {
func (r *githubRepository) createWebhook(ctx context.Context, logger *slog.Logger) (pgh.WebhookConfig, error) {
secret, err := r.secrets.Encrypt(ctx, r.config.Spec.GitHub.Token)
if err != nil {
return pgh.WebhookConfig{}, fmt.Errorf("encrypt webhook secret: %w", err)
}
cfg := pgh.WebhookConfig{
URL: r.config.Spec.GitHub.WebhookURL,
Secret: r.config.Spec.GitHub.WebhookSecret,
URL: r.webhookURL,
Secret: secret,
ContentType: "json",
Events: []string{"push", "pull_request"},
Events: subscribedEvents,
Active: true,
}
if err := r.gh.CreateWebhook(ctx, r.config.Spec.GitHub.Owner, r.config.Spec.GitHub.Repository, cfg); err != nil {
return err
hook, err := r.gh.CreateWebhook(ctx, r.config.Spec.GitHub.Owner, r.config.Spec.GitHub.Repository, cfg)
if err != nil {
return pgh.WebhookConfig{}, err
}
logger.InfoContext(ctx, "webhook created", "url", cfg.URL)
return nil
logger.InfoContext(ctx, "webhook created", "url", cfg.URL, "id", hook.ID)
return hook, nil
}
func (r *githubRepository) updateWebhook(ctx context.Context, logger *slog.Logger, oldRepo *githubRepository) (UndoFunc, error) {
// updateWebhook checks if the webhook needs to be updated and updates it if necessary.
// if the webhook does not exist, it will create it.
func (r *githubRepository) updateWebhook(ctx context.Context, logger *slog.Logger) (pgh.WebhookConfig, bool, error) {
if r.config.Status.Webhook == nil {
hook, err := r.createWebhook(ctx, logger)
if err != nil {
return pgh.WebhookConfig{}, false, err
}
return hook, true, nil
}
owner := r.config.Spec.GitHub.Owner
repoName := r.config.Spec.GitHub.Repository
hooks, err := r.gh.ListWebhooks(ctx, owner, repoName)
if err != nil {
return nil, fmt.Errorf("list existing webhooks: %w", err)
}
newCfg := r.config.Spec.GitHub
oldCfg := oldRepo.Config().Spec.GitHub
hook, err := r.gh.GetWebhook(ctx, owner, repoName, r.config.Status.Webhook.ID)
switch {
case newCfg.WebhookURL != oldCfg.WebhookURL:
// In this case we cannot find out out which webhook to update, so we delete the old one and create a new one
if err := r.createWebhook(ctx, logger); err != nil {
return nil, fmt.Errorf("create new webhook: %w", err)
case errors.Is(err, pgh.ErrResourceNotFound):
hook, err := r.createWebhook(ctx, logger)
if err != nil {
return pgh.WebhookConfig{}, false, err
}
undoFunc := UndoFunc(func(ctx context.Context) error {
if err := r.deleteWebhook(ctx, logger); err != nil {
return fmt.Errorf("revert create new webhook: %w", err)
}
logger.InfoContext(ctx, "create new webhook reverted", "url", newCfg.WebhookURL)
return nil
})
if err := oldRepo.deleteWebhook(ctx, logger); err != nil {
return undoFunc, fmt.Errorf("delete old webhook: %w", err)
}
undoFunc = undoFunc.Chain(ctx, func(ctx context.Context) error {
if err := oldRepo.createWebhook(ctx, logger); err != nil {
return fmt.Errorf("revert delete old webhook: %w", err)
}
logger.InfoContext(ctx, "delete old webhook reverted", "url", oldCfg.WebhookURL)
return nil
})
return undoFunc, nil
case newCfg.WebhookSecret != oldCfg.WebhookSecret:
for _, hook := range hooks {
if hook.URL == oldCfg.WebhookURL {
hook.Secret = newCfg.WebhookSecret
err := r.gh.EditWebhook(ctx, owner, repoName, hook)
if err != nil {
return nil, fmt.Errorf("update webhook secret: %w", err)
}
logger.InfoContext(ctx, "webhook secret updated", "url", newCfg.WebhookURL)
return func(ctx context.Context) error {
hook.Secret = oldCfg.WebhookSecret
if err := r.gh.EditWebhook(ctx, owner, repoName, hook); err != nil {
return fmt.Errorf("revert webhook secret: %w", err)
}
logger.InfoContext(ctx, "webhook secret reverted", "url", oldCfg.WebhookURL)
return nil
}, nil
}
}
return nil, errors.New("webhook not found")
default:
return nil, nil
return hook, true, nil
case err != nil:
return pgh.WebhookConfig{}, false, fmt.Errorf("get webhook: %w", err)
}
var mustUpdate bool
secret, err := r.secrets.Encrypt(ctx, r.config.Spec.GitHub.Token)
if err != nil {
return pgh.WebhookConfig{}, false, fmt.Errorf("encrypt webhook secret: %w", err)
}
// Compare with status secret as we cannot get the screen from the webhook
if secret != r.config.Status.Webhook.Secret {
mustUpdate = true
hook.Secret = r.config.Status.Webhook.Secret
}
if hook.URL != r.config.Status.Webhook.URL {
mustUpdate = true
hook.URL = r.webhookURL
}
if !slices.Equal(hook.Events, subscribedEvents) {
mustUpdate = true
hook.Events = subscribedEvents
}
if !mustUpdate {
return hook, false, nil
}
if err := r.gh.EditWebhook(ctx, owner, repoName, hook); err != nil {
return pgh.WebhookConfig{}, false, fmt.Errorf("edit webhook: %w", err)
}
// HACK: GitHub does not return the secret, so we need to update it manually
hook.Secret = secret
return hook, true, nil
}
func (r *githubRepository) deleteWebhook(ctx context.Context, logger *slog.Logger) error {
if r.config.Status.Webhook == nil {
return fmt.Errorf("webhook not found")
}
owner := r.config.Spec.GitHub.Owner
name := r.config.Spec.GitHub.Repository
id := r.config.Status.Webhook.ID
hooks, err := r.gh.ListWebhooks(ctx, owner, name)
if err != nil {
return fmt.Errorf("list existing webhooks: %w", err)
if err := r.gh.DeleteWebhook(ctx, owner, name, id); err != nil {
return fmt.Errorf("delete webhook: %w", err)
}
for _, hook := range hooks {
if hook.URL == r.config.Spec.GitHub.WebhookURL {
if err := r.gh.DeleteWebhook(ctx, owner, name, hook.ID); err != nil {
return fmt.Errorf("delete webhook: %w", err)
}
}
}
logger.InfoContext(ctx, "webhook deleted", "url", r.config.Spec.GitHub.WebhookURL)
logger.InfoContext(ctx, "webhook deleted", "url", r.config.Status.Webhook.URL, "id", id)
return nil
}
func (r *githubRepository) AfterCreate(ctx context.Context, logger *slog.Logger) error {
return r.createWebhook(ctx, logger)
}
func (r *githubRepository) BeginUpdate(ctx context.Context, logger *slog.Logger, old Repository) (UndoFunc, error) {
oldGitRepo, ok := old.(*githubRepository)
if !ok {
return nil, fmt.Errorf("old repository is not a github repository")
func (r *githubRepository) OnCreate(ctx context.Context, logger *slog.Logger) (*provisioning.RepositoryStatus, error) {
hook, err := r.createWebhook(ctx, logger)
if err != nil {
return nil, err
}
return r.updateWebhook(ctx, logger, oldGitRepo)
status := r.config.Status.DeepCopy()
status.Webhook = &provisioning.WebhookStatus{
ID: hook.ID,
URL: hook.URL,
Secret: hook.Secret,
SubscribedEvents: hook.Events,
}
return status, nil
}
func (r *githubRepository) AfterDelete(ctx context.Context, logger *slog.Logger) error {
func (r *githubRepository) OnUpdate(ctx context.Context, logger *slog.Logger) (*provisioning.RepositoryStatus, error) {
hook, updated, err := r.updateWebhook(ctx, logger)
if err != nil {
return nil, err
}
if !updated {
return nil, nil
}
status := r.config.Status.DeepCopy()
status.Webhook = &provisioning.WebhookStatus{
ID: hook.ID,
URL: hook.URL,
Secret: hook.Secret,
SubscribedEvents: hook.Events,
}
return status, nil
}
func (r *githubRepository) OnDelete(ctx context.Context, logger *slog.Logger) error {
return r.deleteWebhook(ctx, logger)
}
@@ -87,7 +87,8 @@ type Client interface {
GetBranch(ctx context.Context, owner, repository, branchName string) (Branch, error)
ListWebhooks(ctx context.Context, owner, repository string) ([]WebhookConfig, error)
CreateWebhook(ctx context.Context, owner, repository string, cfg WebhookConfig) error
CreateWebhook(ctx context.Context, owner, repository string, cfg WebhookConfig) (WebhookConfig, error)
GetWebhook(ctx context.Context, owner, repository string, webhookID int64) (WebhookConfig, error)
DeleteWebhook(ctx context.Context, owner, repository string, webhookID int64) error
EditWebhook(ctx context.Context, owner, repository string, cfg WebhookConfig) error
@@ -192,21 +192,23 @@ func (_m *MockClient) CreatePullRequestFileComment(ctx context.Context, owner st
}
// CreateWebhook provides a mock function with given fields: ctx, owner, repository, cfg
func (_m *MockClient) CreateWebhook(ctx context.Context, owner string, repository string, cfg WebhookConfig) error {
func (_m *MockClient) CreateWebhook(ctx context.Context, owner string, repository string, cfg WebhookConfig) (WebhookConfig, error) {
ret := _m.Called(ctx, owner, repository, cfg)
if len(ret) == 0 {
panic("no return value specified for CreateWebhook")
}
var r0 error
if rf, ok := ret.Get(0).(func(context.Context, string, string, WebhookConfig) error); ok {
r0 = rf(ctx, owner, repository, cfg)
var r0 WebhookConfig
var r1 error
if rf, ok := ret.Get(0).(func(context.Context, string, string, WebhookConfig) (WebhookConfig, error)); ok {
r0, r1 = rf(ctx, owner, repository, cfg)
} else {
r0 = ret.Error(0)
r0 = ret.Get(0).(WebhookConfig)
r1 = ret.Error(1)
}
return r0
return r0, r1
}
// DeleteFile provides a mock function with given fields: ctx, owner, repository, path, branch, message, hash
@@ -367,6 +369,34 @@ func (_m *MockClient) GetTree(ctx context.Context, owner string, repository stri
return r0, r1, r2
}
// GetWebhook provides a mock function with given fields: ctx, owner, repository, webhookID
func (_m *MockClient) GetWebhook(ctx context.Context, owner string, repository string, webhookID int64) (WebhookConfig, error) {
ret := _m.Called(ctx, owner, repository, webhookID)
if len(ret) == 0 {
panic("no return value specified for GetWebhook")
}
var r0 WebhookConfig
var r1 error
if rf, ok := ret.Get(0).(func(context.Context, string, string, int64) (WebhookConfig, error)); ok {
return rf(ctx, owner, repository, webhookID)
}
if rf, ok := ret.Get(0).(func(context.Context, string, string, int64) WebhookConfig); ok {
r0 = rf(ctx, owner, repository, webhookID)
} else {
r0 = ret.Get(0).(WebhookConfig)
}
if rf, ok := ret.Get(1).(func(context.Context, string, string, int64) error); ok {
r1 = rf(ctx, owner, repository, webhookID)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// IsAuthenticated provides a mock function with given fields: ctx
func (_m *MockClient) IsAuthenticated(ctx context.Context) error {
ret := _m.Called(ctx)
@@ -496,7 +526,8 @@ func (_m *MockClient) UpdateFile(ctx context.Context, owner string, repository s
func NewMockClient(t interface {
mock.TestingT
Cleanup(func())
}) *MockClient {
},
) *MockClient {
mock := &MockClient{}
mock.Mock.Test(t)
@@ -392,7 +392,7 @@ func (r *realImpl) ListWebhooks(ctx context.Context, owner, repository string) (
return ret, nil
}
func (r *realImpl) CreateWebhook(ctx context.Context, owner, repository string, cfg WebhookConfig) error {
func (r *realImpl) CreateWebhook(ctx context.Context, owner, repository string, cfg WebhookConfig) (WebhookConfig, error) {
if cfg.ContentType == "" {
cfg.ContentType = "form"
}
@@ -407,12 +407,51 @@ func (r *realImpl) CreateWebhook(ctx context.Context, owner, repository string,
URL: &cfg.URL,
},
}
_, _, err := r.gh.Repositories.CreateHook(ctx, owner, repository, hook)
createdHook, _, err := r.gh.Repositories.CreateHook(ctx, owner, repository, hook)
var ghErr *github.ErrorResponse
if errors.As(err, &ghErr) && ghErr.Response.StatusCode == http.StatusServiceUnavailable {
return ErrServiceUnavailable
return WebhookConfig{}, ErrServiceUnavailable
}
return err
return WebhookConfig{
ID: createdHook.GetID(),
// events is not returned by GitHub.
Events: cfg.Events,
Active: createdHook.GetActive(),
URL: createdHook.GetConfig().GetURL(),
ContentType: createdHook.GetConfig().GetContentType(),
// Secret is not returned by GitHub.
Secret: cfg.Secret,
}, nil
}
func (r *realImpl) GetWebhook(ctx context.Context, owner, repository string, webhookID int64) (WebhookConfig, error) {
hook, _, err := r.gh.Repositories.GetHook(ctx, owner, repository, webhookID)
if err != nil {
var ghErr *github.ErrorResponse
if errors.As(err, &ghErr) && ghErr.Response.StatusCode == http.StatusServiceUnavailable {
return WebhookConfig{}, ErrServiceUnavailable
}
if ghErr.Response.StatusCode == http.StatusNotFound {
return WebhookConfig{}, ErrResourceNotFound
}
return WebhookConfig{}, err
}
contentType := hook.GetConfig().GetContentType()
if contentType == "" {
contentType = "json"
}
return WebhookConfig{
ID: hook.GetID(),
Events: hook.Events,
Active: hook.GetActive(),
URL: hook.GetConfig().GetURL(),
ContentType: contentType,
// Intentionally not setting Secret.
}, nil
}
func (r *realImpl) DeleteWebhook(ctx context.Context, owner, repository string, webhookID int64) error {
@@ -353,14 +353,14 @@ func (r *localRepository) Webhook(ctx context.Context, logger *slog.Logger, req
}, nil
}
func (r *localRepository) AfterCreate(ctx context.Context, logger *slog.Logger) error {
return nil
}
func (r *localRepository) BeginUpdate(ctx context.Context, logger *slog.Logger, old Repository) (UndoFunc, error) {
func (r *localRepository) OnCreate(ctx context.Context, logger *slog.Logger) (*provisioning.RepositoryStatus, error) {
return nil, nil
}
func (r *localRepository) AfterDelete(ctx context.Context, logger *slog.Logger) error {
func (r *localRepository) OnUpdate(ctx context.Context, logger *slog.Logger) (*provisioning.RepositoryStatus, error) {
return nil, nil
}
func (r *localRepository) OnDelete(ctx context.Context, logger *slog.Logger) error {
return nil
}
@@ -14,19 +14,6 @@ import (
var ErrFileNotFound error = fs.ErrNotExist
// UndoFunc is a function that can be called to undo a previous operation
type UndoFunc func(context.Context) error
// Chain is a helper function to chain UndoFuncs together
func (f UndoFunc) Chain(ctx context.Context, next UndoFunc) UndoFunc {
return func(ctx context.Context) error {
if err := f(ctx); err != nil {
return err
}
return next(ctx)
}
}
type FileInfo struct {
// Path to the file on disk.
// No leading or trailing slashes will be contained within.
@@ -116,9 +103,9 @@ type Repository interface {
Webhook(ctx context.Context, logger *slog.Logger, req *http.Request) (*provisioning.WebhookResponse, error)
// Hooks called after the repository has been created, updated or deleted
AfterCreate(ctx context.Context, logger *slog.Logger) error
BeginUpdate(ctx context.Context, logger *slog.Logger, old Repository) (UndoFunc, error)
AfterDelete(ctx context.Context, logger *slog.Logger) error
OnCreate(ctx context.Context, logger *slog.Logger) (*provisioning.RepositoryStatus, error)
OnUpdate(ctx context.Context, logger *slog.Logger) (*provisioning.RepositoryStatus, error)
OnDelete(ctx context.Context, logger *slog.Logger) error
}
// VersionedRepository is a repository that supports versioning
@@ -117,14 +117,14 @@ func (r *s3Repository) Webhook(ctx context.Context, logger *slog.Logger, req *ht
}
}
func (r *s3Repository) AfterCreate(ctx context.Context, logger *slog.Logger) error {
return nil
func (r *s3Repository) OnCreate(ctx context.Context, logger *slog.Logger) (*provisioning.RepositoryStatus, error) {
return nil, nil
}
func (r *s3Repository) BeginUpdate(ctx context.Context, logger *slog.Logger, old Repository) (UndoFunc, error) {
return func(ctx context.Context) error { return nil }, nil
func (r *s3Repository) OnUpdate(ctx context.Context, logger *slog.Logger) (*provisioning.RepositoryStatus, error) {
return nil, nil
}
func (r *s3Repository) AfterDelete(ctx context.Context, logger *slog.Logger) error {
func (r *s3Repository) OnDelete(ctx context.Context, logger *slog.Logger) error {
return nil
}
@@ -113,14 +113,14 @@ func (r *unknownRepository) Webhook(ctx context.Context, logger *slog.Logger, re
}
}
func (r *unknownRepository) AfterCreate(ctx context.Context, logger *slog.Logger) error {
return nil
}
func (r *unknownRepository) BeginUpdate(ctx context.Context, logger *slog.Logger, old Repository) (UndoFunc, error) {
func (r *unknownRepository) OnCreate(ctx context.Context, logger *slog.Logger) (*provisioning.RepositoryStatus, error) {
return nil, nil
}
func (r *unknownRepository) AfterDelete(ctx context.Context, logger *slog.Logger) error {
func (r *unknownRepository) OnUpdate(ctx context.Context, logger *slog.Logger) (*provisioning.RepositoryStatus, error) {
return nil, nil
}
func (r *unknownRepository) OnDelete(ctx context.Context, logger *slog.Logger) error {
return nil
}
@@ -7,7 +7,7 @@ import (
"time"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/tools/cache"
@@ -19,6 +19,7 @@ import (
informer "github.com/grafana/grafana/pkg/generated/informers/externalversions/provisioning/v0alpha1"
listers "github.com/grafana/grafana/pkg/generated/listers/provisioning/v0alpha1"
"github.com/grafana/grafana/pkg/registry/apis/provisioning/auth"
"github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs"
)
// RepositoryController controls how and when CRD is established.
@@ -27,6 +28,8 @@ type RepositoryController struct {
repoLister listers.RepositoryLister
repoSynced cache.InformerSynced
jobs jobs.JobQueue
// Converts config to instance
repoGetter RepoGetter
identities auth.BackgroundIdentityService
@@ -48,6 +51,7 @@ func NewRepositoryController(
repoGetter RepoGetter,
identities auth.BackgroundIdentityService,
tester *RepositoryTester,
jobs jobs.JobQueue,
) (*RepositoryController, error) {
rc := &RepositoryController{
client: provisioningClient,
@@ -63,6 +67,7 @@ func NewRepositoryController(
identities: identities,
tester: tester,
logger: slog.Default().With("logger", "provisioning-repository-controller"),
jobs: jobs,
}
_, err := repoInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -167,19 +172,35 @@ func (rc *RepositoryController) sync(key string) error {
return err
}
ctx := context.Background()
cachedRepo, err := rc.repoLister.Repositories(namespace).Get(name)
if apierrors.IsNotFound(err) {
rc.logger.DebugContext(ctx, "repository not found", "key", key)
return nil
}
if err != nil {
return err
}
rc.logger.DebugContext(ctx, "cached repository", "key", key, "repo", cachedRepo)
id, err := rc.identities.WorkerIdentity(ctx, cachedRepo.Namespace)
if err != nil {
return err
}
ctx = identity.WithRequester(ctx, id)
logger := rc.logger.With("repository", cachedRepo.Name, "namespace", cachedRepo.Namespace)
repo, err := rc.repoGetter.AsRepository(ctx, cachedRepo)
if err != nil {
return fmt.Errorf("unable to create repository from configuration: %w", err)
}
// The repository is deleted
if cachedRepo.DeletionTimestamp != nil {
if err := rc.processDeletedItem(context.Background(), cachedRepo); err != nil {
return err
}
// FIXME: this is never called because the cache does not contain the repository
logger.InfoContext(ctx, "deleting repository")
return repo.OnDelete(ctx, logger)
}
// Did the spec change
@@ -195,58 +216,69 @@ func (rc *RepositoryController) sync(key string) error {
return nil
}
err = rc.processHealthCheck(context.Background(), cachedRepo)
if apierrors.IsNotFound(err) || apierrors.IsConflict(err) {
// deleted or changed in the meantime, we'll get called again
return nil
}
rc.logger.Info("TODO, whatever cleanup is required for", "repository", cachedRepo.Name)
return err
}
func (rc *RepositoryController) processHealthCheck(ctx context.Context, cachedRepo *provisioning.Repository) error {
// This is used for the health check client
id, err := rc.identities.WorkerIdentity(ctx, cachedRepo.Namespace)
res, err := rc.tester.TestRepository(ctx, repo)
if err != nil {
return err
}
ctx = identity.WithRequester(ctx, id)
// Make a copy we can mutate
cfg := cachedRepo.DeepCopy()
repo, err := rc.repoGetter.AsRepository(ctx, cfg)
if err != nil {
cfg.Status.Health = provisioning.HealthStatus{
Healthy: false,
Message: []string{
"Unable to create repository from configuration",
res = &provisioning.TestResults{
Success: false,
Errors: []string{
"error running test repository",
err.Error(),
},
}
} else {
res, err := rc.tester.TestRepository(ctx, repo)
if err != nil {
res = &provisioning.TestResults{
Success: false,
Errors: []string{
"error running test repository",
err.Error(),
},
}
var status *provisioning.RepositoryStatus
if res.Success {
if cachedRepo.Status.Initialized {
status, err = repo.OnUpdate(ctx, logger)
if err != nil {
return fmt.Errorf("on create repository: %w", err)
}
} else {
status, err = repo.OnCreate(ctx, logger)
if err != nil {
return fmt.Errorf("on create repository: %w", err)
}
}
cfg.Status.Health = provisioning.HealthStatus{
Healthy: res.Success,
Message: res.Errors,
job, err := rc.jobs.Add(ctx, &provisioning.Job{
ObjectMeta: v1.ObjectMeta{
Namespace: cachedRepo.Namespace,
Labels: map[string]string{
"repository": cachedRepo.Name,
},
},
Spec: provisioning.JobSpec{
Action: provisioning.JobActionSync,
},
})
if err != nil {
return fmt.Errorf("trigger sync job: %w", err)
}
logger.InfoContext(ctx, "sync job triggered", "job", job.Name)
} else {
logger.ErrorContext(ctx, "repository is unhealthy", "errors", res.Errors)
}
cfg.Status.Health.Checked = time.Now().UnixMilli()
cfg.Status.Health.Generation = cachedRepo.Generation
_, err = rc.client.Repositories(cfg.GetNamespace()).
UpdateStatus(ctx, cfg, metav1.UpdateOptions{})
return err
}
if status == nil {
status = cachedRepo.Status.DeepCopy()
} else {
status.Initialized = true
}
status.Health = provisioning.HealthStatus{
Checked: status.Health.Checked,
Generation: status.Health.Generation,
Healthy: res.Success,
Message: res.Errors,
}
cfg := cachedRepo.DeepCopy()
cfg.Status = *status
if _, err := rc.client.Repositories(cachedRepo.GetNamespace()).
UpdateStatus(ctx, cfg, v1.UpdateOptions{}); err != nil {
return fmt.Errorf("update status: %w", err)
}
func (rc *RepositoryController) processDeletedItem(ctx context.Context, cachedRepo *provisioning.Repository) error {
return nil
}
@@ -54,22 +54,42 @@ func NewReplicator(
// Sync replicates all files in the repository.
func (r *Replicator) Sync(ctx context.Context) error {
// FIXME: how to handle the scenario in which folder changes?
cfg := r.repository.Config()
lastCommit := cfg.Status.Sync.Hash
versionedRepo, isVersioned := r.repository.(repository.VersionedRepository)
// started := time.Now()
if err := r.ensureRepositoryFolderExists(ctx); err != nil {
return fmt.Errorf("ensure repository folder exists: %w", err)
}
var latest string
if !isVersioned || lastCommit == "" {
switch {
case !isVersioned:
r.logger.InfoContext(ctx, "replicate tree unversioned repository")
if err := r.replicateTree(ctx, ""); err != nil {
return fmt.Errorf("replicate tree: %w", err)
}
} else {
case lastCommit == "":
var err error
latest, err = versionedRepo.LatestRef(ctx, r.logger)
if err != nil {
return fmt.Errorf("latest ref: %w", err)
}
if err := r.replicateTree(ctx, latest); err != nil {
return fmt.Errorf("replicate tree: %w", err)
}
r.logger.InfoContext(ctx, "initial replication for versioned repository", "latest", latest)
default:
var err error
latest, err = versionedRepo.LatestRef(ctx, r.logger)
if err != nil {
return fmt.Errorf("latest ref: %w", err)
}
r.logger.InfoContext(ctx, "replicate changes for versioned repository", "last_commit", lastCommit, "latest", latest)
changes, err := versionedRepo.CompareFiles(ctx, r.logger, lastCommit, latest)
if err != nil {
return fmt.Errorf("compare files: %w", err)
@@ -80,10 +100,12 @@ func (r *Replicator) Sync(ctx context.Context) error {
}
}
// TODO: move the sync status to the job worker
status := &provisioning.SyncStatus{
// TODO: rename to ref
// FIXME: these create infinite loop
// Started: started.Unix(),
// Finished: time.Now().Unix(),
Hash: latest,
// TODO: add timestamp
}
cfg.Status.Sync = *status
@@ -501,6 +523,38 @@ func (r *Replicator) fetchRepoFolderTree(ctx context.Context) (*folderTree, erro
}, nil
}
func (r *Replicator) ensureRepositoryFolderExists(ctx context.Context) error {
_, err := r.folders.Get(ctx, r.repository.Config().Spec.Folder, metav1.GetOptions{})
if err == nil {
return nil
} else if !apierrors.IsNotFound(err) {
return fmt.Errorf("failed to check if folder exists: %w", err)
}
cfg := r.repository.Config()
title := cfg.Spec.Title
if title == "" {
title = cfg.Spec.Folder
}
if _, err := r.folders.Create(ctx, &unstructured.Unstructured{
Object: map[string]any{
"metadata": map[string]any{
"name": cfg.Spec.Folder,
"namespace": cfg.GetNamespace(),
},
"spec": map[string]any{
"title": title,
"description": "Repository-managed folder",
},
},
}, metav1.CreateOptions{}); err != nil {
return fmt.Errorf("failed to create folder: %w", err)
}
return nil
}
func (*Replicator) marshalPreferredFormat(obj any, name string, repo repository.Repository) (body []byte, fileName string, err error) {
if repo.Config().Spec.PreferYAML {
body, err = yaml.Marshal(obj)
@@ -0,0 +1,26 @@
package secrets
import (
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
)
// FIXME: this is a temporary service/package until we can make use of
// the new secrets service in app platform
type Service struct {
encryptionKey []byte
}
func NewService(encryptionKey string) *Service {
return &Service{encryptionKey: []byte(encryptionKey)}
}
func (s *Service) Encrypt(ctx context.Context, data string) (string, error) {
h := hmac.New(sha256.New, s.encryptionKey)
h.Write([]byte(data))
hashed := h.Sum(nil)
return hex.EncodeToString(hashed), nil
}
-27
View File
@@ -3,7 +3,6 @@ package provisioning
import (
"context"
"encoding/json"
"fmt"
"io"
"log/slog"
"net/http"
@@ -13,7 +12,6 @@ import (
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/util/validation/field"
"k8s.io/apiserver/pkg/registry/rest"
@@ -133,31 +131,6 @@ func (t *RepositoryTester) TestRepository(ctx context.Context, repo repository.R
return rsp, nil
}
// Check if the folder exists
cfg := repo.Config()
if cfg.Spec.Folder != "" {
if t.clientFactory == nil {
return nil, fmt.Errorf("client factory is not initialized properly")
}
dynamicClient, _, err := t.clientFactory.New(cfg.Namespace)
if err != nil {
return nil, err
}
folderClient := dynamicClient.Resource(schema.GroupVersionResource{
Group: "folder.grafana.app",
Version: "v0alpha1",
Resource: "folders",
})
_, err = folderClient.Get(ctx, cfg.Spec.Folder, metav1.GetOptions{})
if apierrors.IsNotFound(err) {
return &provisioning.TestResults{
Code: http.StatusFailedDependency,
Success: false,
Errors: []string{"Configured folder not found"},
}, nil
}
}
return repo.Test(ctx, t.logger)
}