-
Notifications
You must be signed in to change notification settings - Fork 30
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Adds initial controller implementation
Signed-off-by: Takeshi Yoneda <t.y.mathetake@gmail.com>
- Loading branch information
Showing
17 changed files
with
1,457 additions
and
50 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,7 +1,54 @@ | ||
| package main | ||
|
|
||
| import "github.com/envoyproxy/ai-gateway/internal/version" | ||
| import ( | ||
| "context" | ||
| "flag" | ||
| "log" | ||
| "log/slog" | ||
| "os" | ||
| "os/signal" | ||
| "syscall" | ||
|
|
||
| "github.com/go-logr/logr" | ||
| "k8s.io/klog/v2" | ||
| ctrl "sigs.k8s.io/controller-runtime" | ||
|
|
||
| "github.com/envoyproxy/ai-gateway/internal/controller" | ||
| ) | ||
|
|
||
| var ( | ||
| logLevel = flag.String("logLevel", "info", "log level") | ||
| extProcImage = flag.String("extprocImage", | ||
| "ghcr.io/envoyproxy/ai-gateway-extproc:latest", "image for the external processor") | ||
| ) | ||
|
|
||
| func main() { | ||
| println(version.Version) | ||
| flag.Parse() | ||
| var level slog.Level | ||
| if err := level.UnmarshalText([]byte(*logLevel)); err != nil { | ||
| log.Fatalf("failed to unmarshal log level: %v", err) | ||
| } | ||
| l := logr.FromSlogHandler(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{ | ||
| Level: level, | ||
| })) | ||
| klog.SetLogger(l) | ||
|
|
||
| k8sConfig, err := ctrl.GetConfig() | ||
| if err != nil { | ||
| log.Fatalf("failed to get k8s config: %v", err) | ||
| } | ||
|
|
||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| signalsChan := make(chan os.Signal, 1) | ||
| signal.Notify(signalsChan, syscall.SIGINT, syscall.SIGTERM) | ||
| go func() { | ||
| <-signalsChan | ||
| cancel() | ||
| }() | ||
|
|
||
| // TODO: starts the extension server? | ||
|
|
||
| if err := controller.StartControllers(ctx, k8sConfig, l, *logLevel, *extProcImage); err != nil { | ||
| log.Fatalf("failed to start controller: %v", err) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,103 @@ | ||
| package controller | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1" | ||
| "github.com/go-logr/logr" | ||
| apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" | ||
| "k8s.io/apimachinery/pkg/runtime" | ||
| utilruntime "k8s.io/apimachinery/pkg/util/runtime" | ||
| "k8s.io/client-go/kubernetes" | ||
| clientgoscheme "k8s.io/client-go/kubernetes/scheme" | ||
| "k8s.io/client-go/rest" | ||
| ctrl "sigs.k8s.io/controller-runtime" | ||
| "sigs.k8s.io/controller-runtime/pkg/client" | ||
| gwapiv1 "sigs.k8s.io/gateway-api/apis/v1" | ||
| gwapiv1b1 "sigs.k8s.io/gateway-api/apis/v1beta1" | ||
|
|
||
| aigv1a1 "github.com/envoyproxy/ai-gateway/api/v1alpha1" | ||
| ) | ||
|
|
||
| var scheme = runtime.NewScheme() | ||
|
|
||
| func init() { | ||
| utilruntime.Must(clientgoscheme.AddToScheme(scheme)) | ||
| utilruntime.Must(aigv1a1.AddToScheme(scheme)) | ||
| utilruntime.Must(apiextensionsv1.AddToScheme(scheme)) | ||
| utilruntime.Must(egv1a1.AddToScheme(scheme)) | ||
| utilruntime.Must(gwapiv1.Install(scheme)) | ||
| utilruntime.Must(gwapiv1b1.Install(scheme)) | ||
| } | ||
|
|
||
| func newClients(config *rest.Config) (kubeClient client.Client, kube kubernetes.Interface, err error) { | ||
| kubeClient, err = client.New(config, client.Options{Scheme: scheme}) | ||
| if err != nil { | ||
| return nil, nil, fmt.Errorf("failed to create new client: %w", err) | ||
| } | ||
|
|
||
| kube, err = kubernetes.NewForConfig(config) | ||
| if err != nil { | ||
| return nil, nil, fmt.Errorf("failed to create kubernetes client: %w", err) | ||
| } | ||
| return kubeClient, kube, nil | ||
| } | ||
|
|
||
| // StartControllers starts the controllers for the AI Gateway. | ||
| // This blocks until the manager is stopped. | ||
| func StartControllers(ctx context.Context, config *rest.Config, logger logr.Logger, logLevel string, extProcImage string) error { | ||
| mgr, err := ctrl.NewManager(config, ctrl.Options{ | ||
| Scheme: scheme, | ||
| LeaderElection: true, | ||
| LeaderElectionID: "envoy-ai-gateway-controller", | ||
| }) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to create new controller manager: %w", err) | ||
| } | ||
|
|
||
| clientForRouteC, kubeForRouteC, err := newClients(config) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to create new clients: %w", err) | ||
| } | ||
|
|
||
| sinkChan := make(chan configSinkEvent, 100) | ||
| routeC := newLLMRouteController(clientForRouteC, kubeForRouteC, logger, logLevel, extProcImage, sinkChan) | ||
| if err = ctrl.NewControllerManagedBy(mgr). | ||
| For(&aigv1a1.LLMRoute{}). | ||
| Complete(routeC); err != nil { | ||
| return fmt.Errorf("failed to create controller for LLMRoute: %w", err) | ||
| } | ||
|
|
||
| clientForBackendC, kubeForBackendC, err := newClients(config) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to create new clients: %w", err) | ||
| } | ||
|
|
||
| backendC := newLLMBackendController(clientForBackendC, kubeForBackendC, logger, sinkChan) | ||
| if err = ctrl.NewControllerManagedBy(mgr). | ||
| For(&aigv1a1.LLMBackend{}). | ||
| Complete(backendC); err != nil { | ||
| return fmt.Errorf("failed to create controller for LLMBackend: %w", err) | ||
| } | ||
|
|
||
| clientForConfigSink, kubeForConfigSink, err := newClients(config) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to create new clients: %w", err) | ||
| } | ||
|
|
||
| sink := newConfigSink(clientForConfigSink, kubeForConfigSink, logger, sinkChan) | ||
|
|
||
| // Wait for the manager to become the leader before starting the controllers. | ||
| <-mgr.Elected() | ||
|
|
||
| // Before starting the manager, initialize the config sink to sync all LLMBackend and LLMRoute objects in the cluster. | ||
| if err = sink.init(ctx); err != nil { | ||
| return fmt.Errorf("failed to initialize config sink: %w", err) | ||
| } | ||
|
|
||
| if err = mgr.Start(ctx); err != nil { // This blocks until the manager is stopped. | ||
| return fmt.Errorf("failed to start controller manager: %w", err) | ||
| } | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| package controller | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/go-logr/logr" | ||
| "k8s.io/client-go/kubernetes" | ||
| ctrl "sigs.k8s.io/controller-runtime" | ||
| "sigs.k8s.io/controller-runtime/pkg/client" | ||
| "sigs.k8s.io/controller-runtime/pkg/reconcile" | ||
|
|
||
| aigv1a1 "github.com/envoyproxy/ai-gateway/api/v1alpha1" | ||
| ) | ||
|
|
||
| // llmBackendController implements [reconcile.TypedReconciler] for [aigv1a1.LLMBackend]. | ||
| // | ||
| // This handles the LLMBackend resource and sends it to the config sink so that it can modify the configuration together with the state of other resources. | ||
| type llmBackendController struct { | ||
| client client.Client | ||
| kube kubernetes.Interface | ||
| logger logr.Logger | ||
| eventChan chan configSinkEvent | ||
| } | ||
|
|
||
| func newLLMBackendController(client client.Client, kube kubernetes.Interface, logger logr.Logger, ch chan configSinkEvent) *llmBackendController { | ||
| return &llmBackendController{ | ||
| client: client, | ||
| kube: kube, | ||
| logger: logger, | ||
| eventChan: ch, | ||
| } | ||
| } | ||
|
|
||
| // Reconcile implements the [reconcile.TypedReconciler] for [aigv1a1.LLMBackend]. | ||
| func (l *llmBackendController) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) { | ||
| var llmBackend aigv1a1.LLMBackend | ||
| if err := l.client.Get(ctx, req.NamespacedName, &llmBackend); err != nil { | ||
| if client.IgnoreNotFound(err) == nil { | ||
| l.eventChan <- configSinkEventLLMBackendDeleted{namespace: req.Namespace, name: req.Name} | ||
| ctrl.Log.Info("Deleting LLMBackend", | ||
| "namespace", req.Namespace, "name", req.Name) | ||
| return ctrl.Result{}, nil | ||
| } | ||
| return ctrl.Result{}, err | ||
| } | ||
| // Send the LLMBackend to the config sink so that it can modify the configuration together with the state of other resources. | ||
| l.eventChan <- llmBackend.DeepCopy() | ||
| return ctrl.Result{}, nil | ||
| } |
Oops, something went wrong.