feat: sqs-operator — CRD QueueService, контроллер, ElasticMQ per-tenant

This commit is contained in:
Naeel
2026-04-07 13:56:53 +03:00
parent a22a37bef3
commit cbe61d9f62
9 changed files with 1956 additions and 0 deletions
+90
View File
@@ -0,0 +1,90 @@
// Создан: 2026-04-07
// main.go — точка входа SQS Operator.
// Запускает controller-runtime manager с QueueServiceReconciler.
// Отдельный оператор — независим от sless-operator (отдельная репа в будущем).
package main
import (
"flag"
"os"
"k8s.io/apimachinery/pkg/runtime"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
clientgoscheme "k8s.io/client-go/kubernetes/scheme"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/healthz"
"sigs.k8s.io/controller-runtime/pkg/log/zap"
sqsv1alpha1 "gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sqs-operator/api/v1alpha1"
"gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sqs-operator/controllers"
"gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sqs-operator/internal/config"
)
var scheme = runtime.NewScheme()
func init() {
utilruntime.Must(clientgoscheme.AddToScheme(scheme))
utilruntime.Must(sqsv1alpha1.AddToScheme(scheme))
}
func main() {
var metricsAddr string
var enableLeaderElection bool
var probeAddr string
flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "metrics endpoint address")
flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "health probe endpoint address")
flag.BoolVar(&enableLeaderElection, "leader-elect", false, "enable leader election")
opts := zap.Options{Development: true}
opts.BindFlags(flag.CommandLine)
flag.Parse()
ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
log := ctrl.Log.WithName("sqs-operator")
cfg, err := config.Load()
if err != nil {
log.Error(err, "load config")
os.Exit(1)
}
mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
Scheme: scheme,
MetricsBindAddress: metricsAddr,
Port: 9443,
HealthProbeBindAddress: probeAddr,
LeaderElection: enableLeaderElection,
LeaderElectionID: "sqs-operator.kube5s.ru",
})
if err != nil {
log.Error(err, "unable to start manager")
os.Exit(1)
}
if err = (&controllers.QueueServiceReconciler{
Client: mgr.GetClient(),
Scheme: mgr.GetScheme(),
SQSExternalHost: cfg.SQSExternalHost,
ElasticMQImage: cfg.ElasticMQImage,
}).SetupWithManager(mgr); err != nil {
log.Error(err, "unable to create controller", "controller", "QueueService")
os.Exit(1)
}
if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
log.Error(err, "unable to set up health check")
os.Exit(1)
}
if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
log.Error(err, "unable to set up ready check")
os.Exit(1)
}
log.Info("starting SQS operator", "externalHost", cfg.SQSExternalHost)
if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
log.Error(err, "problem running manager")
os.Exit(1)
}
}