mirror of
https://github.com/imjasonh/git-k8s
synced 2026-07-18 15:05:45 +00:00
Implement Git-as-K8s control plane with Knative controllers
Build a complete Kubernetes control plane for Git operations using Knative's pkg framework, go-git for in-memory Git operations, and ko for builds. Architecture: - 4 CRDs: GitRepository, GitBranch, GitPushTransaction, GitRepoSync - Push controller: atomic Git push via go-git with CAS (compare-and-swap) - Sync controller: two-way sync with in-memory merge base calculation - Resolver controller: automated 3-way merge for conflicted syncs - In-cluster Gitea server for backing storage - Full RBAC, ko-based deployments, and code-gen scaffolding Note: go.sum needs generation via `go mod tidy` in an environment with full internet access (knative.dev/pkg v0.0.0-20250520014526-44579e9ce5ed). https://claude.ai/code/session_01QfFzqKQUsxxBsiZUhuJ3pG
This commit is contained in:
Claude
commit
8e65104201
27 changed files with 3198 additions and 0 deletions
102
pkg/reconciler/sync/controller.go
Normal file
102
pkg/reconciler/sync/controller.go
Normal file
|
|
@ -0,0 +1,102 @@
|
|||
package sync
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/client-go/dynamic"
|
||||
"k8s.io/client-go/dynamic/dynamicinformer"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
"knative.dev/pkg/configmap"
|
||||
"knative.dev/pkg/controller"
|
||||
"knative.dev/pkg/injection"
|
||||
"knative.dev/pkg/logging"
|
||||
|
||||
gitv1alpha1 "github.com/imjasonh/git-k8s/pkg/apis/git/v1alpha1"
|
||||
gitclient "github.com/imjasonh/git-k8s/pkg/client"
|
||||
)
|
||||
|
||||
var (
|
||||
repoSyncGVR = schema.GroupVersionResource{
|
||||
Group: gitv1alpha1.GroupName,
|
||||
Version: gitv1alpha1.Version,
|
||||
Resource: "gitreposyncs",
|
||||
}
|
||||
branchGVR = schema.GroupVersionResource{
|
||||
Group: gitv1alpha1.GroupName,
|
||||
Version: gitv1alpha1.Version,
|
||||
Resource: "gitbranches",
|
||||
}
|
||||
)
|
||||
|
||||
// NewController creates a new controller for GitRepoSync resources.
|
||||
func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
|
||||
logger := logging.FromContext(ctx)
|
||||
|
||||
cfg := injection.GetConfig(ctx)
|
||||
dynClient, err := dynamic.NewForConfig(cfg)
|
||||
if err != nil {
|
||||
logger.Fatalf("Error building dynamic client: %v", err)
|
||||
}
|
||||
|
||||
gitClient := gitclient.NewFromDynamic(dynClient)
|
||||
|
||||
factory := dynamicinformer.NewDynamicSharedInformerFactory(dynClient, 30*time.Second)
|
||||
syncInformer := factory.ForResource(repoSyncGVR)
|
||||
branchInformer := factory.ForResource(branchGVR)
|
||||
|
||||
r := &Reconciler{
|
||||
dynamicClient: dynClient,
|
||||
gitClient: gitClient,
|
||||
}
|
||||
|
||||
impl := controller.NewContext(ctx, r, controller.ControllerOptions{
|
||||
WorkQueueName: "GitRepoSyncs",
|
||||
Logger: logger,
|
||||
})
|
||||
|
||||
logger.Info("Setting up event handlers for GitRepoSync")
|
||||
|
||||
// Watch GitRepoSync resources directly.
|
||||
syncInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: func(obj interface{}) {
|
||||
impl.Enqueue(obj)
|
||||
},
|
||||
UpdateFunc: func(oldObj, newObj interface{}) {
|
||||
impl.Enqueue(newObj)
|
||||
},
|
||||
})
|
||||
|
||||
// Watch GitBranch resources and enqueue the owning GitRepoSync.
|
||||
// This ensures syncs are re-evaluated when branches change.
|
||||
branchInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
|
||||
FilterFunc: func(obj interface{}) bool {
|
||||
// Only process branches that have an owner reference to a GitRepoSync.
|
||||
u, ok := obj.(*unstructured.Unstructured)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
for _, ref := range u.GetOwnerReferences() {
|
||||
if ref.Kind == "GitRepoSync" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
},
|
||||
Handler: cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: func(obj interface{}) {
|
||||
impl.Enqueue(obj)
|
||||
},
|
||||
UpdateFunc: func(oldObj, newObj interface{}) {
|
||||
impl.Enqueue(newObj)
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
factory.Start(ctx.Done())
|
||||
factory.WaitForCacheSync(ctx.Done())
|
||||
|
||||
return impl
|
||||
}
|
||||
228
pkg/reconciler/sync/sync.go
Normal file
228
pkg/reconciler/sync/sync.go
Normal file
|
|
@ -0,0 +1,228 @@
|
|||
package sync
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/go-git/go-git/v5"
|
||||
"github.com/go-git/go-git/v5/plumbing"
|
||||
"github.com/go-git/go-git/v5/storage/memory"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/client-go/dynamic"
|
||||
"knative.dev/pkg/logging"
|
||||
"knative.dev/pkg/reconciler"
|
||||
|
||||
gitv1alpha1 "github.com/imjasonh/git-k8s/pkg/apis/git/v1alpha1"
|
||||
gitclient "github.com/imjasonh/git-k8s/pkg/client"
|
||||
)
|
||||
|
||||
// Reconciler implements the reconcile logic for GitRepoSync.
|
||||
type Reconciler struct {
|
||||
dynamicClient dynamic.Interface
|
||||
gitClient *gitclient.GitV1alpha1Client
|
||||
}
|
||||
|
||||
// Reconcile implements the controller.Reconciler interface.
|
||||
func (r *Reconciler) Reconcile(ctx context.Context, key string) error {
|
||||
logger := logging.FromContext(ctx)
|
||||
|
||||
namespace, name, err := splitKey(key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Fetch the GitRepoSync resource.
|
||||
syncObj, err := r.gitClient.GitRepoSyncs(namespace).Get(ctx, name, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
return fmt.Errorf("getting GitRepoSync %s/%s: %w", namespace, name, err)
|
||||
}
|
||||
|
||||
// Get the branches for both repos.
|
||||
branchA, err := r.findBranch(ctx, namespace, syncObj.Spec.RepoA.Name, syncObj.Spec.BranchName)
|
||||
if err != nil {
|
||||
return r.updateSyncStatus(ctx, syncObj, gitv1alpha1.SyncPhaseConflicted,
|
||||
fmt.Sprintf("finding branch for repo A: %v", err), "", "", "")
|
||||
}
|
||||
|
||||
branchB, err := r.findBranch(ctx, namespace, syncObj.Spec.RepoB.Name, syncObj.Spec.BranchName)
|
||||
if err != nil {
|
||||
return r.updateSyncStatus(ctx, syncObj, gitv1alpha1.SyncPhaseConflicted,
|
||||
fmt.Sprintf("finding branch for repo B: %v", err), "", "", "")
|
||||
}
|
||||
|
||||
commitA := branchA.Status.HeadCommit
|
||||
commitB := branchB.Status.HeadCommit
|
||||
|
||||
// If both branches point to the same commit, they're in sync.
|
||||
if commitA == commitB {
|
||||
logger.Infof("GitRepoSync %s/%s: repos are in sync at %s", namespace, name, commitA)
|
||||
return r.updateSyncStatus(ctx, syncObj, gitv1alpha1.SyncPhaseInSync, "Repos are in sync", commitA, commitB, commitA)
|
||||
}
|
||||
|
||||
// Repos differ - calculate merge base to determine which is ahead.
|
||||
logger.Infof("GitRepoSync %s/%s: commits differ (A=%s, B=%s), calculating merge base", namespace, name, commitA, commitB)
|
||||
|
||||
// Fetch repo A to get commit objects.
|
||||
repoASpec, err := r.gitClient.GitRepositories(namespace).Get(ctx, syncObj.Spec.RepoA.Name, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
return fmt.Errorf("getting repo A: %w", err)
|
||||
}
|
||||
|
||||
mergeBase, err := r.calculateMergeBase(ctx, repoASpec.Spec.URL, commitA, commitB)
|
||||
if err != nil {
|
||||
return r.updateSyncStatus(ctx, syncObj, gitv1alpha1.SyncPhaseConflicted,
|
||||
fmt.Sprintf("calculating merge base: %v", err), commitA, commitB, "")
|
||||
}
|
||||
|
||||
switch {
|
||||
case mergeBase == commitA:
|
||||
// B is ahead of A. Create a push transaction to update A.
|
||||
logger.Infof("Repo B is ahead. Creating push transaction to update A.")
|
||||
if err := r.createPushTransaction(ctx, namespace, syncObj, syncObj.Spec.RepoA.Name, commitB, syncObj.Spec.BranchName, commitA); err != nil {
|
||||
return fmt.Errorf("creating push transaction for A: %w", err)
|
||||
}
|
||||
return r.updateSyncStatus(ctx, syncObj, gitv1alpha1.SyncPhaseSyncing,
|
||||
"Repo B is ahead; pushing to A", commitA, commitB, mergeBase)
|
||||
|
||||
case mergeBase == commitB:
|
||||
// A is ahead of B. Create a push transaction to update B.
|
||||
logger.Infof("Repo A is ahead. Creating push transaction to update B.")
|
||||
if err := r.createPushTransaction(ctx, namespace, syncObj, syncObj.Spec.RepoB.Name, commitA, syncObj.Spec.BranchName, commitB); err != nil {
|
||||
return fmt.Errorf("creating push transaction for B: %w", err)
|
||||
}
|
||||
return r.updateSyncStatus(ctx, syncObj, gitv1alpha1.SyncPhaseSyncing,
|
||||
"Repo A is ahead; pushing to B", commitA, commitB, mergeBase)
|
||||
|
||||
default:
|
||||
// Diverged - both repos have changes since the merge base.
|
||||
logger.Infof("Repos have diverged (mergeBase=%s). Marking as conflicted.", mergeBase)
|
||||
return r.updateSyncStatus(ctx, syncObj, gitv1alpha1.SyncPhaseConflicted,
|
||||
fmt.Sprintf("Repos have diverged from merge base %s", mergeBase), commitA, commitB, mergeBase)
|
||||
}
|
||||
}
|
||||
|
||||
// findBranch finds a GitBranch matching the given repository and branch name.
|
||||
func (r *Reconciler) findBranch(ctx context.Context, namespace, repoName, branchName string) (*gitv1alpha1.GitBranch, error) {
|
||||
branches, err := r.gitClient.GitBranches(namespace).List(ctx, metav1.ListOptions{})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("listing branches: %w", err)
|
||||
}
|
||||
for i := range branches.Items {
|
||||
b := &branches.Items[i]
|
||||
if b.Spec.RepositoryRef == repoName && b.Spec.BranchName == branchName {
|
||||
return b, nil
|
||||
}
|
||||
}
|
||||
return nil, fmt.Errorf("branch %q not found for repo %q", branchName, repoName)
|
||||
}
|
||||
|
||||
// calculateMergeBase computes the merge base between two commits entirely in memory.
|
||||
// It clones the repository into memory, then walks the commit history to find the
|
||||
// common ancestor.
|
||||
func (r *Reconciler) calculateMergeBase(ctx context.Context, repoURL, commitAHash, commitBHash string) (string, error) {
|
||||
// Clone into memory.
|
||||
storer := memory.NewStorage()
|
||||
repo, err := git.Clone(storer, nil, &git.CloneOptions{
|
||||
URL: repoURL,
|
||||
})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("cloning for merge base: %w", err)
|
||||
}
|
||||
|
||||
hashA := plumbing.NewHash(commitAHash)
|
||||
hashB := plumbing.NewHash(commitBHash)
|
||||
|
||||
commitA, err := repo.CommitObject(hashA)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("getting commit A (%s): %w", commitAHash, err)
|
||||
}
|
||||
|
||||
commitB, err := repo.CommitObject(hashB)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("getting commit B (%s): %w", commitBHash, err)
|
||||
}
|
||||
|
||||
// Use go-git's merge base calculation.
|
||||
bases, err := commitA.MergeBase(commitB)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("calculating merge base: %w", err)
|
||||
}
|
||||
|
||||
if len(bases) == 0 {
|
||||
return "", fmt.Errorf("no common ancestor found between %s and %s", commitAHash, commitBHash)
|
||||
}
|
||||
|
||||
return bases[0].Hash.String(), nil
|
||||
}
|
||||
|
||||
// createPushTransaction creates a GitPushTransaction to push a commit to a target repo's branch.
|
||||
func (r *Reconciler) createPushTransaction(ctx context.Context, namespace string, syncObj *gitv1alpha1.GitRepoSync, targetRepo, sourceCommit, branchName, expectedOld string) error {
|
||||
txn := &gitv1alpha1.GitPushTransaction{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
GenerateName: fmt.Sprintf("%s-sync-", syncObj.Name),
|
||||
Namespace: namespace,
|
||||
Labels: map[string]string{
|
||||
"git.k8s.io/repo-sync": syncObj.Name,
|
||||
"git.k8s.io/target": targetRepo,
|
||||
},
|
||||
OwnerReferences: []metav1.OwnerReference{
|
||||
{
|
||||
APIVersion: gitv1alpha1.SchemeGroupVersion.String(),
|
||||
Kind: "GitRepoSync",
|
||||
Name: syncObj.Name,
|
||||
UID: syncObj.UID,
|
||||
},
|
||||
},
|
||||
},
|
||||
Spec: gitv1alpha1.GitPushTransactionSpec{
|
||||
RepositoryRef: targetRepo,
|
||||
Atomic: true,
|
||||
RefSpecs: []gitv1alpha1.PushRefSpec{
|
||||
{
|
||||
Source: sourceCommit,
|
||||
Destination: fmt.Sprintf("refs/heads/%s", branchName),
|
||||
ExpectedOldCommit: expectedOld,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
_, err := r.gitClient.GitPushTransactions(namespace).Create(ctx, txn, metav1.CreateOptions{})
|
||||
return err
|
||||
}
|
||||
|
||||
// updateSyncStatus updates the GitRepoSync status fields.
|
||||
func (r *Reconciler) updateSyncStatus(ctx context.Context, syncObj *gitv1alpha1.GitRepoSync, phase gitv1alpha1.SyncPhase, message, commitA, commitB, mergeBase string) error {
|
||||
syncObj.Status.Phase = phase
|
||||
syncObj.Status.Message = message
|
||||
syncObj.Status.RepoACommit = commitA
|
||||
syncObj.Status.RepoBCommit = commitB
|
||||
syncObj.Status.MergeBase = mergeBase
|
||||
|
||||
if phase == gitv1alpha1.SyncPhaseInSync {
|
||||
now := metav1.Now()
|
||||
syncObj.Status.LastSyncTime = &now
|
||||
}
|
||||
|
||||
_, err := r.gitClient.GitRepoSyncs(syncObj.Namespace).UpdateStatus(ctx, syncObj, metav1.UpdateOptions{})
|
||||
return err
|
||||
}
|
||||
|
||||
func splitKey(key string) (string, string, error) {
|
||||
for i := range key {
|
||||
if key[i] == '/' {
|
||||
return key[:i], key[i+1:], nil
|
||||
}
|
||||
}
|
||||
return "", key, nil
|
||||
}
|
||||
|
||||
// Ensure Reconciler implements the reconciler interface.
|
||||
var _ reconciler.LeaderAware = (*Reconciler)(nil)
|
||||
|
||||
func (r *Reconciler) Promote(bkt reconciler.Bucket, enq func(bkt reconciler.Bucket, key string) error) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Reconciler) Demote(bkt reconciler.Bucket) {
|
||||
}
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue