deploy: EMQX, iot-operator, managed Postgres, Docker Hub image
- Добавлен EMQX deployment + WS ingress (адаптирован из sless) - Добавлен iot-operator deployment с RBAC (ServiceAccount, ClusterRole) - Postgres: заменён self-hosted на managed (dc5db45d namespace) - Image: naeel/iot-operator:v0.2.0 (Docker Hub) - Убран imagePullSecrets (Docker Hub публичный)
This commit is contained in:
Naeel
+115
-115
@@ -13,131 +13,131 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
"time"
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"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"
|
||||
"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"
|
||||
|
||||
iotv1alpha1 "gitea.services.ngcloud.ru/Nail/IoT/api/v1alpha1"
|
||||
iotcontrollers "gitea.services.ngcloud.ru/Nail/IoT/controllers"
|
||||
iotapi "gitea.services.ngcloud.ru/Nail/IoT/internal/api"
|
||||
"gitea.services.ngcloud.ru/Nail/IoT/internal/api/handler"
|
||||
"gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg"
|
||||
iotv1alpha1 "gitea.services.ngcloud.ru/Nail/IoT/api/v1alpha1"
|
||||
iotcontrollers "gitea.services.ngcloud.ru/Nail/IoT/controllers"
|
||||
iotapi "gitea.services.ngcloud.ru/Nail/IoT/internal/api"
|
||||
"gitea.services.ngcloud.ru/Nail/IoT/internal/api/handler"
|
||||
"gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg"
|
||||
)
|
||||
|
||||
var scheme = runtime.NewScheme()
|
||||
|
||||
func init() {
|
||||
utilruntime.Must(clientgoscheme.AddToScheme(scheme))
|
||||
// IoT API group iot.kube5s.ru/v1alpha1
|
||||
utilruntime.Must(iotv1alpha1.AddToScheme(scheme))
|
||||
utilruntime.Must(clientgoscheme.AddToScheme(scheme))
|
||||
// IoT API group iot.kube5s.ru/v1alpha1
|
||||
utilruntime.Must(iotv1alpha1.AddToScheme(scheme))
|
||||
}
|
||||
|
||||
func main() {
|
||||
log := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
|
||||
ctrl.SetLogger(zap.New(zap.UseDevMode(true)))
|
||||
log := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
|
||||
ctrl.SetLogger(zap.New(zap.UseDevMode(true)))
|
||||
|
||||
log.Info("starting IoT managed service operator")
|
||||
log.Info("starting IoT managed service operator")
|
||||
|
||||
// Controller Manager — управляет reconcile loop для IoTDevice CRD
|
||||
mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
|
||||
Scheme: scheme,
|
||||
HealthProbeBindAddress: ":8081",
|
||||
LeaderElection: false,
|
||||
})
|
||||
if err != nil {
|
||||
log.Error("unable to create controller manager", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// Регистрируем IoTDevice controller
|
||||
if err = (&iotcontrollers.IoTDeviceReconciler{
|
||||
Client: mgr.GetClient(),
|
||||
Scheme: mgr.GetScheme(),
|
||||
}).SetupWithManager(mgr); err != nil {
|
||||
log.Error("unable to create IoTDevice controller", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// Health/ready пробы для k8s
|
||||
if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
|
||||
log.Error("unable to set up health check", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
|
||||
log.Error("unable to set up ready check", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// IoT Postgres — опционален (если IOT_PG_DSN не задан — телеметрия отключена)
|
||||
iotPGStore, err := iotpg.NewFromEnv(log)
|
||||
if err != nil {
|
||||
log.Error("failed to init IoT Postgres", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
if iotPGStore != nil {
|
||||
defer iotPGStore.Close()
|
||||
}
|
||||
|
||||
// REST API handler
|
||||
h := &handler.Handler{
|
||||
K8s: mgr.GetClient(),
|
||||
Scheme: mgr.GetScheme(),
|
||||
IoTPG: iotPGStore,
|
||||
Log: log,
|
||||
}
|
||||
|
||||
router := iotapi.NewRouter(h, log)
|
||||
|
||||
// HTTP API сервер на порту 9090
|
||||
apiServer := &http.Server{
|
||||
Addr: ":9090",
|
||||
Handler: router,
|
||||
ReadTimeout: 30 * time.Second,
|
||||
WriteTimeout: 60 * time.Second,
|
||||
}
|
||||
|
||||
// Запускаем API сервер в горутине
|
||||
go func() {
|
||||
log.Info("starting REST API server", "addr", ":9090")
|
||||
if err := apiServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
||||
log.Error("API server failed", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}()
|
||||
|
||||
// Graceful shutdown
|
||||
ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
|
||||
defer cancel()
|
||||
|
||||
// Запускаем controller-manager (блокирует до ctx.Done)
|
||||
go func() {
|
||||
log.Info("starting controller manager")
|
||||
if err := mgr.Start(ctx); err != nil {
|
||||
log.Error("controller manager failed", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}()
|
||||
|
||||
<-ctx.Done()
|
||||
log.Info("shutting down IoT operator")
|
||||
|
||||
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer shutdownCancel()
|
||||
if err := apiServer.Shutdown(shutdownCtx); err != nil {
|
||||
log.Error("API server shutdown error", "err", err)
|
||||
}
|
||||
|
||||
fmt.Println("IoT operator stopped")
|
||||
// Controller Manager — управляет reconcile loop для IoTDevice CRD
|
||||
mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
|
||||
Scheme: scheme,
|
||||
HealthProbeBindAddress: ":8081",
|
||||
LeaderElection: false,
|
||||
})
|
||||
if err != nil {
|
||||
log.Error("unable to create controller manager", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// Регистрируем IoTDevice controller
|
||||
if err = (&iotcontrollers.IoTDeviceReconciler{
|
||||
Client: mgr.GetClient(),
|
||||
Scheme: mgr.GetScheme(),
|
||||
}).SetupWithManager(mgr); err != nil {
|
||||
log.Error("unable to create IoTDevice controller", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// Health/ready пробы для k8s
|
||||
if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
|
||||
log.Error("unable to set up health check", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
|
||||
log.Error("unable to set up ready check", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// IoT Postgres — опционален (если IOT_PG_DSN не задан — телеметрия отключена)
|
||||
iotPGStore, err := iotpg.NewFromEnv(log)
|
||||
if err != nil {
|
||||
log.Error("failed to init IoT Postgres", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
if iotPGStore != nil {
|
||||
defer iotPGStore.Close()
|
||||
}
|
||||
|
||||
// REST API handler
|
||||
h := &handler.Handler{
|
||||
K8s: mgr.GetClient(),
|
||||
Scheme: mgr.GetScheme(),
|
||||
IoTPG: iotPGStore,
|
||||
Log: log,
|
||||
}
|
||||
|
||||
router := iotapi.NewRouter(h, log)
|
||||
|
||||
// HTTP API сервер на порту 9090
|
||||
apiServer := &http.Server{
|
||||
Addr: ":9090",
|
||||
Handler: router,
|
||||
ReadTimeout: 30 * time.Second,
|
||||
WriteTimeout: 60 * time.Second,
|
||||
}
|
||||
|
||||
// Запускаем API сервер в горутине
|
||||
go func() {
|
||||
log.Info("starting REST API server", "addr", ":9090")
|
||||
if err := apiServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
||||
log.Error("API server failed", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}()
|
||||
|
||||
// Graceful shutdown
|
||||
ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
|
||||
defer cancel()
|
||||
|
||||
// Запускаем controller-manager (блокирует до ctx.Done)
|
||||
go func() {
|
||||
log.Info("starting controller manager")
|
||||
if err := mgr.Start(ctx); err != nil {
|
||||
log.Error("controller manager failed", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}()
|
||||
|
||||
<-ctx.Done()
|
||||
log.Info("shutting down IoT operator")
|
||||
|
||||
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer shutdownCancel()
|
||||
if err := apiServer.Shutdown(shutdownCtx); err != nil {
|
||||
log.Error("API server shutdown error", "err", err)
|
||||
}
|
||||
|
||||
fmt.Println("IoT operator stopped")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user