feat: proxy /fn/{namespace}/{name} — обход wildcard DNS
Проблема: wildcard DNS *.fn.kube5s.ru недоступен.
Решение: прокси через sless-api.kube5s.ru/fn/{ns}/{name}.
- handler/invoke.go: прокси к Service функции внутри кластера
- router.go: /fn/ без auth токена, /v1/ с auth (gorilla Use())
- config.go: поле ExternalURL (EXTERNAL_URL env)
- trigger_controller.go: если ExternalURL задан — URL = ExternalURL/fn/{ns}/{fn}
иначе fallback: Ingress + поддомен (прежнее поведение)
- operator.yaml: EXTERNAL_URL=https://sless-api.kube5s.ru, image v0.1.5
Оператор v0.1.5 задеплоен.
E2E: curl https://sless-api.kube5s.ru/fn/default/hello-node → {"message":"Hello, Naeel! (nodejs20)"}
This commit is contained in:
+219
-210
@@ -6,31 +6,32 @@
|
|||||||
package controllers
|
package controllers
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
|
||||||
batchv1 "k8s.io/api/batch/v1"
|
batchv1 "k8s.io/api/batch/v1"
|
||||||
corev1 "k8s.io/api/core/v1"
|
corev1 "k8s.io/api/core/v1"
|
||||||
netv1 "k8s.io/api/networking/v1"
|
netv1 "k8s.io/api/networking/v1"
|
||||||
"k8s.io/apimachinery/pkg/api/errors"
|
"k8s.io/apimachinery/pkg/api/errors"
|
||||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
"k8s.io/apimachinery/pkg/runtime"
|
"k8s.io/apimachinery/pkg/runtime"
|
||||||
ctrl "sigs.k8s.io/controller-runtime"
|
ctrl "sigs.k8s.io/controller-runtime"
|
||||||
"sigs.k8s.io/controller-runtime/pkg/client"
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||||
"sigs.k8s.io/controller-runtime/pkg/log"
|
"sigs.k8s.io/controller-runtime/pkg/log"
|
||||||
|
|
||||||
slessv1alpha1 "gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/api/v1alpha1"
|
slessv1alpha1 "gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/api/v1alpha1"
|
||||||
)
|
)
|
||||||
|
|
||||||
// TriggerReconciler reconciles a Trigger object
|
// TriggerReconciler reconciles a Trigger object
|
||||||
type TriggerReconciler struct {
|
type TriggerReconciler struct {
|
||||||
client.Client
|
client.Client
|
||||||
Scheme *runtime.Scheme
|
Scheme *runtime.Scheme
|
||||||
IngressHost string // базовый домен для http триггеров, например: fn.kube5s.ru
|
IngressHost string // базовый домен для http триггеров (fallback): fn.kube5s.ru
|
||||||
|
// ExternalURL — если задан, URL функции = ExternalURL/fn/{namespace}/{name}
|
||||||
|
// (обходит wildcard DNS, трафик идёт через sless-api.kube5s.ru)
|
||||||
|
ExternalURL string
|
||||||
}
|
}
|
||||||
|
|
||||||
//+kubebuilder:rbac:groups=sless.kube5s.ru,resources=triggers,verbs=get;list;watch;create;update;patch;delete
|
|
||||||
//+kubebuilder:rbac:groups=sless.kube5s.ru,resources=triggers/status,verbs=get;update;patch
|
|
||||||
//+kubebuilder:rbac:groups=sless.kube5s.ru,resources=triggers/finalizers,verbs=update
|
//+kubebuilder:rbac:groups=sless.kube5s.ru,resources=triggers/finalizers,verbs=update
|
||||||
//+kubebuilder:rbac:groups=networking.k8s.io,resources=ingresses,verbs=get;list;watch;create;update;patch;delete
|
//+kubebuilder:rbac:groups=networking.k8s.io,resources=ingresses,verbs=get;list;watch;create;update;patch;delete
|
||||||
//+kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch;create;update;patch;delete
|
//+kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch;create;update;patch;delete
|
||||||
@@ -40,221 +41,229 @@ const triggerFinalizer = "sless.kube5s.ru/trigger-finalizer"
|
|||||||
|
|
||||||
// Reconcile — управляет ресурсами для триггера в зависимости от его типа.
|
// Reconcile — управляет ресурсами для триггера в зависимости от его типа.
|
||||||
func (r *TriggerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
|
func (r *TriggerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
|
||||||
logger := log.FromContext(ctx)
|
logger := log.FromContext(ctx)
|
||||||
|
|
||||||
tr := &slessv1alpha1.Trigger{}
|
tr := &slessv1alpha1.Trigger{}
|
||||||
if err := r.Get(ctx, req.NamespacedName, tr); err != nil {
|
if err := r.Get(ctx, req.NamespacedName, tr); err != nil {
|
||||||
if errors.IsNotFound(err) {
|
if errors.IsNotFound(err) {
|
||||||
return ctrl.Result{}, nil
|
return ctrl.Result{}, nil
|
||||||
}
|
}
|
||||||
return ctrl.Result{}, fmt.Errorf("get trigger: %w", err)
|
return ctrl.Result{}, fmt.Errorf("get trigger: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Обработка удаления
|
||||||
|
if !tr.DeletionTimestamp.IsZero() {
|
||||||
|
return r.handleTriggerDeletion(ctx, tr)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Добавляем finalizer если ещё нет
|
||||||
|
if !containsString(tr.Finalizers, triggerFinalizer) {
|
||||||
|
tr.Finalizers = append(tr.Finalizers, triggerFinalizer)
|
||||||
|
if err := r.Update(ctx, tr); err != nil {
|
||||||
|
return ctrl.Result{}, fmt.Errorf("add trigger finalizer: %w", err)
|
||||||
|
}
|
||||||
|
return ctrl.Result{Requeue: true}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Проверяем что Function существует и готова
|
||||||
|
fn := &slessv1alpha1.Function{}
|
||||||
|
if err := r.Get(ctx, client.ObjectKey{Name: tr.Spec.FunctionRef, Namespace: tr.Namespace}, fn); err != nil {
|
||||||
|
if errors.IsNotFound(err) {
|
||||||
|
tr.Status.Active = false
|
||||||
|
tr.Status.Message = "function not found: " + tr.Spec.FunctionRef
|
||||||
|
_ = r.Status().Update(ctx, tr)
|
||||||
|
return ctrl.Result{}, nil
|
||||||
|
}
|
||||||
|
return ctrl.Result{}, fmt.Errorf("get function: %w", err)
|
||||||
|
}
|
||||||
|
if fn.Status.Phase != slessv1alpha1.FunctionPhaseReady {
|
||||||
|
tr.Status.Active = false
|
||||||
|
tr.Status.Message = "waiting for function to be Ready (current: " + string(fn.Status.Phase) + ")"
|
||||||
|
_ = r.Status().Update(ctx, tr)
|
||||||
|
// Повторный reconcile придёт когда Function изменится (watch ниже)
|
||||||
|
return ctrl.Result{}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
switch tr.Spec.Type {
|
||||||
|
case slessv1alpha1.TriggerTypeHTTP:
|
||||||
|
logger.Info("reconcile http trigger", "trigger", tr.Name)
|
||||||
|
return r.reconcileHTTP(ctx, tr, fn)
|
||||||
|
case slessv1alpha1.TriggerTypeCron:
|
||||||
|
logger.Info("reconcile cron trigger", "trigger", tr.Name)
|
||||||
|
return r.reconcileCron(ctx, tr, fn)
|
||||||
|
}
|
||||||
|
|
||||||
|
return ctrl.Result{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Обработка удаления
|
// reconcileHTTP создаёт Service для HTTP триггера.
|
||||||
if !tr.DeletionTimestamp.IsZero() {
|
// Если ExternalURL задан — URL = {ExternalURL}/fn/{namespace}/{name} (прокси через API).
|
||||||
return r.handleTriggerDeletion(ctx, tr)
|
// Иначе (fallback) — URL = https://{funcName}-{namespace}.{IngressHost} + создаёт Ingress.
|
||||||
}
|
|
||||||
|
|
||||||
// Добавляем finalizer если ещё нет
|
|
||||||
if !containsString(tr.Finalizers, triggerFinalizer) {
|
|
||||||
tr.Finalizers = append(tr.Finalizers, triggerFinalizer)
|
|
||||||
if err := r.Update(ctx, tr); err != nil {
|
|
||||||
return ctrl.Result{}, fmt.Errorf("add trigger finalizer: %w", err)
|
|
||||||
}
|
|
||||||
return ctrl.Result{Requeue: true}, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Проверяем что Function существует и готова
|
|
||||||
fn := &slessv1alpha1.Function{}
|
|
||||||
if err := r.Get(ctx, client.ObjectKey{Name: tr.Spec.FunctionRef, Namespace: tr.Namespace}, fn); err != nil {
|
|
||||||
if errors.IsNotFound(err) {
|
|
||||||
tr.Status.Active = false
|
|
||||||
tr.Status.Message = "function not found: " + tr.Spec.FunctionRef
|
|
||||||
_ = r.Status().Update(ctx, tr)
|
|
||||||
return ctrl.Result{}, nil
|
|
||||||
}
|
|
||||||
return ctrl.Result{}, fmt.Errorf("get function: %w", err)
|
|
||||||
}
|
|
||||||
if fn.Status.Phase != slessv1alpha1.FunctionPhaseReady {
|
|
||||||
tr.Status.Active = false
|
|
||||||
tr.Status.Message = "waiting for function to be Ready (current: " + string(fn.Status.Phase) + ")"
|
|
||||||
_ = r.Status().Update(ctx, tr)
|
|
||||||
// Повторный reconcile придёт когда Function изменится (watch ниже)
|
|
||||||
return ctrl.Result{}, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
switch tr.Spec.Type {
|
|
||||||
case slessv1alpha1.TriggerTypeHTTP:
|
|
||||||
logger.Info("reconcile http trigger", "trigger", tr.Name)
|
|
||||||
return r.reconcileHTTP(ctx, tr, fn)
|
|
||||||
case slessv1alpha1.TriggerTypeCron:
|
|
||||||
logger.Info("reconcile cron trigger", "trigger", tr.Name)
|
|
||||||
return r.reconcileCron(ctx, tr, fn)
|
|
||||||
}
|
|
||||||
|
|
||||||
return ctrl.Result{}, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// reconcileHTTP создаёт Service и Ingress для HTTP триггера.
|
|
||||||
// URL функции: https://{funcName}-{namespace}.{IngressHost}
|
|
||||||
func (r *TriggerReconciler) reconcileHTTP(ctx context.Context, tr *slessv1alpha1.Trigger, fn *slessv1alpha1.Function) (ctrl.Result, error) {
|
func (r *TriggerReconciler) reconcileHTTP(ctx context.Context, tr *slessv1alpha1.Trigger, fn *slessv1alpha1.Function) (ctrl.Result, error) {
|
||||||
deployNS := "sless-fn-" + tr.Namespace
|
deployNS := "sless-fn-" + tr.Namespace
|
||||||
host := fmt.Sprintf("%s-%s.%s", fn.Name, tr.Namespace, r.IngressHost)
|
|
||||||
pathType := netv1.PathTypePrefix
|
|
||||||
|
|
||||||
// Service — направляет трафик к Deployment функции
|
// Service — направляет трафик к Deployment функции (нужен и для прокси, и для Ingress)
|
||||||
wantSvc := &corev1.Service{
|
wantSvc := &corev1.Service{
|
||||||
ObjectMeta: metav1.ObjectMeta{Name: fn.Name, Namespace: deployNS},
|
ObjectMeta: metav1.ObjectMeta{Name: fn.Name, Namespace: deployNS},
|
||||||
Spec: corev1.ServiceSpec{
|
Spec: corev1.ServiceSpec{
|
||||||
Selector: map[string]string{"app": fn.Name},
|
Selector: map[string]string{"app": fn.Name},
|
||||||
Ports: []corev1.ServicePort{{Port: 8080, Protocol: corev1.ProtocolTCP}},
|
Ports: []corev1.ServicePort{{Port: 8080, Protocol: corev1.ProtocolTCP}},
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
existingSvc := &corev1.Service{}
|
existingSvc := &corev1.Service{}
|
||||||
if err := r.Get(ctx, client.ObjectKey{Name: fn.Name, Namespace: deployNS}, existingSvc); err != nil {
|
if err := r.Get(ctx, client.ObjectKey{Name: fn.Name, Namespace: deployNS}, existingSvc); err != nil {
|
||||||
if errors.IsNotFound(err) {
|
if errors.IsNotFound(err) {
|
||||||
if err := r.Create(ctx, wantSvc); err != nil {
|
if err := r.Create(ctx, wantSvc); err != nil {
|
||||||
return ctrl.Result{}, fmt.Errorf("create service: %w", err)
|
return ctrl.Result{}, fmt.Errorf("create service: %w", err)
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
return ctrl.Result{}, fmt.Errorf("get service: %w", err)
|
return ctrl.Result{}, fmt.Errorf("get service: %w", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Ingress — внешний доступ по host
|
var funcURL string
|
||||||
wantIng := &netv1.Ingress{
|
if r.ExternalURL != "" {
|
||||||
ObjectMeta: metav1.ObjectMeta{
|
// Режим прокси через API сервер — не нужен wildcard DNS
|
||||||
Name: fn.Name,
|
funcURL = fmt.Sprintf("%s/fn/%s/%s", r.ExternalURL, tr.Namespace, fn.Name)
|
||||||
Namespace: deployNS,
|
} else {
|
||||||
Annotations: map[string]string{
|
// Fallback: Ingress с поддоменом (требует wildcard DNS *.IngressHost)
|
||||||
"kubernetes.io/ingress.class": "nginx",
|
host := fmt.Sprintf("%s-%s.%s", fn.Name, tr.Namespace, r.IngressHost)
|
||||||
},
|
pathType := netv1.PathTypePrefix
|
||||||
},
|
wantIng := &netv1.Ingress{
|
||||||
Spec: netv1.IngressSpec{
|
ObjectMeta: metav1.ObjectMeta{
|
||||||
Rules: []netv1.IngressRule{
|
Name: fn.Name,
|
||||||
{
|
Namespace: deployNS,
|
||||||
Host: host,
|
Annotations: map[string]string{
|
||||||
IngressRuleValue: netv1.IngressRuleValue{
|
"kubernetes.io/ingress.class": "nginx",
|
||||||
HTTP: &netv1.HTTPIngressRuleValue{
|
},
|
||||||
Paths: []netv1.HTTPIngressPath{
|
},
|
||||||
{
|
Spec: netv1.IngressSpec{
|
||||||
Path: "/",
|
Rules: []netv1.IngressRule{
|
||||||
PathType: &pathType,
|
{
|
||||||
Backend: netv1.IngressBackend{
|
Host: host,
|
||||||
Service: &netv1.IngressServiceBackend{
|
IngressRuleValue: netv1.IngressRuleValue{
|
||||||
Name: fn.Name,
|
HTTP: &netv1.HTTPIngressRuleValue{
|
||||||
Port: netv1.ServiceBackendPort{Number: 8080},
|
Paths: []netv1.HTTPIngressPath{
|
||||||
},
|
{
|
||||||
},
|
Path: "/",
|
||||||
},
|
PathType: &pathType,
|
||||||
},
|
Backend: netv1.IngressBackend{
|
||||||
},
|
Service: &netv1.IngressServiceBackend{
|
||||||
},
|
Name: fn.Name,
|
||||||
},
|
Port: netv1.ServiceBackendPort{Number: 8080},
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
}
|
},
|
||||||
existingIng := &netv1.Ingress{}
|
},
|
||||||
if err := r.Get(ctx, client.ObjectKey{Name: fn.Name, Namespace: deployNS}, existingIng); err != nil {
|
},
|
||||||
if errors.IsNotFound(err) {
|
},
|
||||||
if err := r.Create(ctx, wantIng); err != nil {
|
},
|
||||||
return ctrl.Result{}, fmt.Errorf("create ingress: %w", err)
|
},
|
||||||
}
|
},
|
||||||
} else {
|
}
|
||||||
return ctrl.Result{}, fmt.Errorf("get ingress: %w", err)
|
existingIng := &netv1.Ingress{}
|
||||||
}
|
if err := r.Get(ctx, client.ObjectKey{Name: fn.Name, Namespace: deployNS}, existingIng); err != nil {
|
||||||
}
|
if errors.IsNotFound(err) {
|
||||||
|
if err := r.Create(ctx, wantIng); err != nil {
|
||||||
|
return ctrl.Result{}, fmt.Errorf("create ingress: %w", err)
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
return ctrl.Result{}, fmt.Errorf("get ingress: %w", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
funcURL = "https://" + host
|
||||||
|
}
|
||||||
|
|
||||||
tr.Status.Active = true
|
tr.Status.Active = true
|
||||||
tr.Status.URL = "https://" + host
|
tr.Status.URL = funcURL
|
||||||
tr.Status.Message = ""
|
tr.Status.Message = ""
|
||||||
if err := r.Status().Update(ctx, tr); err != nil {
|
if err := r.Status().Update(ctx, tr); err != nil {
|
||||||
return ctrl.Result{}, fmt.Errorf("update trigger status: %w", err)
|
return ctrl.Result{}, fmt.Errorf("update trigger status: %w", err)
|
||||||
}
|
}
|
||||||
return ctrl.Result{}, nil
|
return ctrl.Result{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// reconcileCron создаёт CronJob который вызывает функцию по HTTP внутри кластера.
|
// reconcileCron создаёт CronJob который вызывает функцию по HTTP внутри кластера.
|
||||||
// curl делает POST на внутренний Service функции — это исключает внешний round-trip.
|
// curl делает POST на внутренний Service функции — это исключает внешний round-trip.
|
||||||
func (r *TriggerReconciler) reconcileCron(ctx context.Context, tr *slessv1alpha1.Trigger, fn *slessv1alpha1.Function) (ctrl.Result, error) {
|
func (r *TriggerReconciler) reconcileCron(ctx context.Context, tr *slessv1alpha1.Trigger, fn *slessv1alpha1.Function) (ctrl.Result, error) {
|
||||||
deployNS := "sless-fn-" + tr.Namespace
|
deployNS := "sless-fn-" + tr.Namespace
|
||||||
// Внутренний URL: Service должен быть создан HTTP триггером или заранее
|
// Внутренний URL: Service должен быть создан HTTP триггером или заранее
|
||||||
funcURL := fmt.Sprintf("http://%s.%s.svc.cluster.local:8080", fn.Name, deployNS)
|
funcURL := fmt.Sprintf("http://%s.%s.svc.cluster.local:8080", fn.Name, deployNS)
|
||||||
|
|
||||||
wantCJ := &batchv1.CronJob{
|
wantCJ := &batchv1.CronJob{
|
||||||
ObjectMeta: metav1.ObjectMeta{
|
ObjectMeta: metav1.ObjectMeta{
|
||||||
Name: tr.Name,
|
Name: tr.Name,
|
||||||
Namespace: tr.Namespace,
|
Namespace: tr.Namespace,
|
||||||
Labels: map[string]string{"managed-by": "sless", "trigger": tr.Name},
|
Labels: map[string]string{"managed-by": "sless", "trigger": tr.Name},
|
||||||
},
|
},
|
||||||
Spec: batchv1.CronJobSpec{
|
Spec: batchv1.CronJobSpec{
|
||||||
Schedule: tr.Spec.Schedule,
|
Schedule: tr.Spec.Schedule,
|
||||||
JobTemplate: batchv1.JobTemplateSpec{
|
JobTemplate: batchv1.JobTemplateSpec{
|
||||||
Spec: batchv1.JobSpec{
|
Spec: batchv1.JobSpec{
|
||||||
Template: corev1.PodTemplateSpec{
|
Template: corev1.PodTemplateSpec{
|
||||||
Spec: corev1.PodSpec{
|
Spec: corev1.PodSpec{
|
||||||
RestartPolicy: corev1.RestartPolicyOnFailure,
|
RestartPolicy: corev1.RestartPolicyOnFailure,
|
||||||
Containers: []corev1.Container{
|
Containers: []corev1.Container{
|
||||||
{
|
{
|
||||||
// curlimages/curl вызывает функцию по внутреннему адресу
|
// curlimages/curl вызывает функцию по внутреннему адресу
|
||||||
Name: "invoker",
|
Name: "invoker",
|
||||||
Image: "curlimages/curl:latest",
|
Image: "curlimages/curl:latest",
|
||||||
Command: []string{"curl", "-sf", "-X", "POST", funcURL},
|
Command: []string{"curl", "-sf", "-X", "POST", funcURL},
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
existing := &batchv1.CronJob{}
|
existing := &batchv1.CronJob{}
|
||||||
if err := r.Get(ctx, client.ObjectKey{Name: tr.Name, Namespace: tr.Namespace}, existing); err != nil {
|
if err := r.Get(ctx, client.ObjectKey{Name: tr.Name, Namespace: tr.Namespace}, existing); err != nil {
|
||||||
if errors.IsNotFound(err) {
|
if errors.IsNotFound(err) {
|
||||||
if err := r.Create(ctx, wantCJ); err != nil {
|
if err := r.Create(ctx, wantCJ); err != nil {
|
||||||
return ctrl.Result{}, fmt.Errorf("create cronjob: %w", err)
|
return ctrl.Result{}, fmt.Errorf("create cronjob: %w", err)
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
return ctrl.Result{}, fmt.Errorf("get cronjob: %w", err)
|
return ctrl.Result{}, fmt.Errorf("get cronjob: %w", err)
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
// Синхронизируем расписание если оно изменилось
|
// Синхронизируем расписание если оно изменилось
|
||||||
existing.Spec.Schedule = tr.Spec.Schedule
|
existing.Spec.Schedule = tr.Spec.Schedule
|
||||||
if err := r.Update(ctx, existing); err != nil {
|
if err := r.Update(ctx, existing); err != nil {
|
||||||
return ctrl.Result{}, fmt.Errorf("update cronjob schedule: %w", err)
|
return ctrl.Result{}, fmt.Errorf("update cronjob schedule: %w", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
tr.Status.Active = true
|
tr.Status.Active = true
|
||||||
tr.Status.Message = ""
|
tr.Status.Message = ""
|
||||||
if err := r.Status().Update(ctx, tr); err != nil {
|
if err := r.Status().Update(ctx, tr); err != nil {
|
||||||
return ctrl.Result{}, fmt.Errorf("update trigger status: %w", err)
|
return ctrl.Result{}, fmt.Errorf("update trigger status: %w", err)
|
||||||
}
|
}
|
||||||
return ctrl.Result{}, nil
|
return ctrl.Result{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// handleTriggerDeletion удаляет ресурсы триггера и убирает finalizer.
|
// handleTriggerDeletion удаляет ресурсы триггера и убирает finalizer.
|
||||||
func (r *TriggerReconciler) handleTriggerDeletion(ctx context.Context, tr *slessv1alpha1.Trigger) (ctrl.Result, error) {
|
func (r *TriggerReconciler) handleTriggerDeletion(ctx context.Context, tr *slessv1alpha1.Trigger) (ctrl.Result, error) {
|
||||||
if tr.Spec.Type == slessv1alpha1.TriggerTypeCron {
|
if tr.Spec.Type == slessv1alpha1.TriggerTypeCron {
|
||||||
cj := &batchv1.CronJob{}
|
cj := &batchv1.CronJob{}
|
||||||
if err := r.Get(ctx, client.ObjectKey{Name: tr.Name, Namespace: tr.Namespace}, cj); err == nil {
|
if err := r.Get(ctx, client.ObjectKey{Name: tr.Name, Namespace: tr.Namespace}, cj); err == nil {
|
||||||
_ = r.Delete(ctx, cj)
|
_ = r.Delete(ctx, cj)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// Ingress/Service для HTTP триггеров оставляем — они могут быть нужны другим триггерам,
|
// Ingress/Service для HTTP триггеров оставляем — они могут быть нужны другим триггерам,
|
||||||
// полная очистка происходит при удалении Function через function_controller finalizer
|
// полная очистка происходит при удалении Function через function_controller finalizer
|
||||||
tr.Finalizers = removeString(tr.Finalizers, triggerFinalizer)
|
tr.Finalizers = removeString(tr.Finalizers, triggerFinalizer)
|
||||||
if err := r.Update(ctx, tr); err != nil {
|
if err := r.Update(ctx, tr); err != nil {
|
||||||
return ctrl.Result{}, fmt.Errorf("remove trigger finalizer: %w", err)
|
return ctrl.Result{}, fmt.Errorf("remove trigger finalizer: %w", err)
|
||||||
}
|
}
|
||||||
return ctrl.Result{}, nil
|
return ctrl.Result{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// SetupWithManager регистрирует контроллер и настраивает watch на Function.
|
// SetupWithManager регистрирует контроллер и настраивает watch на Function.
|
||||||
// Когда Function переходит в Ready — Trigger автоматически пересчитывается.
|
// Когда Function переходит в Ready — Trigger автоматически пересчитывается.
|
||||||
func (r *TriggerReconciler) SetupWithManager(mgr ctrl.Manager) error {
|
func (r *TriggerReconciler) SetupWithManager(mgr ctrl.Manager) error {
|
||||||
return ctrl.NewControllerManagedBy(mgr).
|
return ctrl.NewControllerManagedBy(mgr).
|
||||||
For(&slessv1alpha1.Trigger{}).
|
For(&slessv1alpha1.Trigger{}).
|
||||||
Complete(r)
|
Complete(r)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -31,6 +31,9 @@ data:
|
|||||||
REGISTRY_SECRET: "sless-registry-auth"
|
REGISTRY_SECRET: "sless-registry-auth"
|
||||||
API_PORT: "9090"
|
API_PORT: "9090"
|
||||||
INGRESS_HOST: "fn.kube5s.ru"
|
INGRESS_HOST: "fn.kube5s.ru"
|
||||||
|
# EXTERNAL_URL — если задан, URL функции = EXTERNAL_URL/fn/{namespace}/{name}
|
||||||
|
# Позволяет обойтись без wildcard DNS *.fn.kube5s.ru
|
||||||
|
EXTERNAL_URL: "https://sless-api.kube5s.ru"
|
||||||
---
|
---
|
||||||
# Secret создаётся отдельно через kubectl (не коммитить секреты в git!)
|
# Secret создаётся отдельно через kubectl (не коммитить секреты в git!)
|
||||||
# Описание ключей:
|
# Описание ключей:
|
||||||
@@ -67,7 +70,7 @@ spec:
|
|||||||
containers:
|
containers:
|
||||||
- name: operator
|
- name: operator
|
||||||
# При обновлении версии оператора — менять тег здесь (не latest!)
|
# При обновлении версии оператора — менять тег здесь (не latest!)
|
||||||
image: naeel/sless-operator:v0.1.4
|
image: naeel/sless-operator:v0.1.5
|
||||||
# Always — чтобы всегда тянуть по точному тегу (не кешировать старый)
|
# Always — чтобы всегда тянуть по точному тегу (не кешировать старый)
|
||||||
imagePullPolicy: Always
|
imagePullPolicy: Always
|
||||||
ports:
|
ports:
|
||||||
|
|||||||
@@ -0,0 +1,65 @@
|
|||||||
|
// Изменено: 2026-03-07
|
||||||
|
// invoke.go — прокси-обработчик для вызова HTTP-триггеров функций.
|
||||||
|
// Маршрут: ANY /fn/{namespace}/{name}
|
||||||
|
// Не защищён auth-токеном — это публичный эндпоинт для вызова функций.
|
||||||
|
// Проксирует запрос к ClusterIP Service функции внутри кластера:
|
||||||
|
// http://{name}.sless-fn-{namespace}.svc.cluster.local:8080
|
||||||
|
// Функция должна быть задеплоена (HTTP триггер активен).
|
||||||
|
|
||||||
|
package handler
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"net/http"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/gorilla/mux"
|
||||||
|
)
|
||||||
|
|
||||||
|
// httpClient используется для обращения к функциям внутри кластера.
|
||||||
|
// Таймаут 30s — достаточно для холодного старта функции.
|
||||||
|
var httpClient = &http.Client{Timeout: 30 * time.Second}
|
||||||
|
|
||||||
|
// InvokeFunction проксирует входящий запрос к Service функции в кластере.
|
||||||
|
// Namespace выбирается из пути, имя функции — тоже из пути.
|
||||||
|
// Сохраняет метод, тело, и Content-Type заголовок оригинального запроса.
|
||||||
|
func (h *Handler) InvokeFunction(w http.ResponseWriter, r *http.Request) {
|
||||||
|
vars := mux.Vars(r)
|
||||||
|
ns := vars["namespace"]
|
||||||
|
name := vars["name"]
|
||||||
|
|
||||||
|
// Внутренний URL к Service функции (DNS внутри кластера)
|
||||||
|
// Namespace функций: sless-fn-{user-namespace}
|
||||||
|
target := fmt.Sprintf("http://%s.sless-fn-%s.svc.cluster.local:8080", name, ns)
|
||||||
|
|
||||||
|
// Создаём проксируемый запрос с тем же методом и телом
|
||||||
|
proxyReq, err := http.NewRequestWithContext(r.Context(), r.Method, target, r.Body)
|
||||||
|
if err != nil {
|
||||||
|
h.Log.Error("invoke: failed to create proxy request", "err", err, "ns", ns, "fn", name)
|
||||||
|
writeJSON(w, http.StatusInternalServerError, errResp("failed to create proxy request"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Пробрасываем Content-Type если есть
|
||||||
|
if ct := r.Header.Get("Content-Type"); ct != "" {
|
||||||
|
proxyReq.Header.Set("Content-Type", ct)
|
||||||
|
}
|
||||||
|
|
||||||
|
resp, err := httpClient.Do(proxyReq)
|
||||||
|
if err != nil {
|
||||||
|
h.Log.Error("invoke: function unreachable", "err", err, "ns", ns, "fn", name, "target", target)
|
||||||
|
writeJSON(w, http.StatusBadGateway, errResp("function unreachable: "+err.Error()))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
|
||||||
|
// Копируем заголовки и статус из ответа функции
|
||||||
|
for k, vals := range resp.Header {
|
||||||
|
for _, v := range vals {
|
||||||
|
w.Header().Add(k, v)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
w.WriteHeader(resp.StatusCode)
|
||||||
|
_, _ = io.Copy(w, resp.Body)
|
||||||
|
}
|
||||||
+13
-6
@@ -17,9 +17,14 @@ import (
|
|||||||
|
|
||||||
// NewRouter собирает gorilla/mux роутер со всеми маршрутами.
|
// NewRouter собирает gorilla/mux роутер со всеми маршрутами.
|
||||||
// apiToken — статический Bearer-токен для v1 аутентификации.
|
// apiToken — статический Bearer-токен для v1 аутентификации.
|
||||||
|
// /fn/{namespace}/{name} — публичный прокси для вызова функций, без auth.
|
||||||
func NewRouter(h *handler.Handler, apiToken string, log *slog.Logger) http.Handler {
|
func NewRouter(h *handler.Handler, apiToken string, log *slog.Logger) http.Handler {
|
||||||
r := mux.NewRouter()
|
r := mux.NewRouter()
|
||||||
|
|
||||||
|
// Публичный прокси для вызова HTTP-триггеров — без auth токена
|
||||||
|
// Все HTTP методы разрешены (GET/POST/PUT/... — решает сама функция)
|
||||||
|
r.PathPrefix("/fn/{namespace}/{name}").HandlerFunc(h.InvokeFunction)
|
||||||
|
|
||||||
// Суброутер для /v1 — все маршруты API
|
// Суброутер для /v1 — все маршруты API
|
||||||
v1 := r.PathPrefix("/v1").Subrouter()
|
v1 := r.PathPrefix("/v1").Subrouter()
|
||||||
|
|
||||||
@@ -47,10 +52,12 @@ func NewRouter(h *handler.Handler, apiToken string, log *slog.Logger) http.Handl
|
|||||||
v1.HandleFunc("/namespaces/{namespace}/jobs/{name}", h.GetJob).Methods(http.MethodGet)
|
v1.HandleFunc("/namespaces/{namespace}/jobs/{name}", h.GetJob).Methods(http.MethodGet)
|
||||||
v1.HandleFunc("/namespaces/{namespace}/jobs/{name}", h.DeleteJob).Methods(http.MethodDelete)
|
v1.HandleFunc("/namespaces/{namespace}/jobs/{name}", h.DeleteJob).Methods(http.MethodDelete)
|
||||||
|
|
||||||
// Цепочка middleware: logging → auth → router
|
// Цепочка middleware: logging → (auth только для /v1/) → router
|
||||||
// Порядок важен: сначала логируем (чтобы видеть все запросы включая отклонённые),
|
// /fn/ — без auth, /v1/ — с auth.
|
||||||
// затем проверяем авторизацию.
|
// Используем gorilla/mux Use() чтобы auth применялся только к v1 суброутеру.
|
||||||
return middleware.Logging(log,
|
v1.Use(func(next http.Handler) http.Handler {
|
||||||
middleware.Auth(apiToken, log, r),
|
return middleware.Auth(apiToken, log, next)
|
||||||
)
|
})
|
||||||
|
|
||||||
|
return middleware.Logging(log, r)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -41,10 +41,15 @@ type Config struct {
|
|||||||
// BuilderImage — образ для сборки функций (kaniko или buildah)
|
// BuilderImage — образ для сборки функций (kaniko или buildah)
|
||||||
BuilderImage string
|
BuilderImage string
|
||||||
|
|
||||||
// IngressHost — базовый домен для HTTP триггеров, например: fn.kube5s.ru
|
// IngressHost — базовый домен для HTTP триггеров (используется только
|
||||||
// URL функции будет: https://{funcName}-{namespace}.{IngressHost}
|
// если ExternalURL не задан, для обратной совместимости).
|
||||||
IngressHost string
|
IngressHost string
|
||||||
|
|
||||||
|
// ExternalURL — публичный URL API сервера, например: https://sless-api.kube5s.ru
|
||||||
|
// Если задан, URL функции формируется как: {ExternalURL}/fn/{namespace}/{name}
|
||||||
|
// Это позволяет обойтись без wildcard DNS *.fn.
|
||||||
|
ExternalURL string
|
||||||
|
|
||||||
// APIToken — статический Bearer-токен для v1 REST API аутентификации.
|
// APIToken — статический Bearer-токен для v1 REST API аутентификации.
|
||||||
// В prod заменить на вызов auth-сервиса.
|
// В prod заменить на вызов auth-сервиса.
|
||||||
APIToken string
|
APIToken string
|
||||||
@@ -117,12 +122,15 @@ func Load() (*Config, error) {
|
|||||||
cfg.BuilderImage = v
|
cfg.BuilderImage = v
|
||||||
}
|
}
|
||||||
|
|
||||||
// IngressHost — базовый домен для HTTP триггеров
|
// IngressHost — базовый домен для HTTP триггеров (fallback)
|
||||||
cfg.IngressHost = os.Getenv("INGRESS_HOST")
|
cfg.IngressHost = os.Getenv("INGRESS_HOST")
|
||||||
if cfg.IngressHost == "" {
|
if cfg.IngressHost == "" {
|
||||||
cfg.IngressHost = "fn.kube5s.ru"
|
cfg.IngressHost = "fn.kube5s.ru"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ExternalURL — публичный URL сервиса для формирования URL функций
|
||||||
|
cfg.ExternalURL = os.Getenv("EXTERNAL_URL")
|
||||||
|
|
||||||
// APIToken — обязательный токен для REST API
|
// APIToken — обязательный токен для REST API
|
||||||
cfg.APIToken = os.Getenv("SLESS_API_TOKEN")
|
cfg.APIToken = os.Getenv("SLESS_API_TOKEN")
|
||||||
if cfg.APIToken == "" {
|
if cfg.APIToken == "" {
|
||||||
|
|||||||
@@ -140,6 +140,7 @@ func main() {
|
|||||||
Client: mgr.GetClient(),
|
Client: mgr.GetClient(),
|
||||||
Scheme: mgr.GetScheme(),
|
Scheme: mgr.GetScheme(),
|
||||||
IngressHost: cfg.IngressHost,
|
IngressHost: cfg.IngressHost,
|
||||||
|
ExternalURL: cfg.ExternalURL,
|
||||||
}).SetupWithManager(mgr); err != nil {
|
}).SetupWithManager(mgr); err != nil {
|
||||||
log.Error("unable to create controller", "controller", "Trigger", "err", err)
|
log.Error("unable to create controller", "controller", "Trigger", "err", err)
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
|
|||||||
Reference in New Issue
Block a user