500 lines
19 KiB
Go
500 lines
19 KiB
Go
// Создано: 2026-03-20
|
||
// ServiceReconciler — контроллер для Service CRD (sless_service).
|
||
// Service = долгоживущий HTTP-сервис с постоянным URL.
|
||
// Lifecycle: Pending → Building (kaniko) → Ready (Deployment+k8s Service+Ingress, URL в Status) / Failed
|
||
//
|
||
// Отличие от Function:
|
||
// Function = oneshot, запускается k8s Job через FunctionJob/invoke.
|
||
// Service = HTTP-сервис, Deployment всегда запущен, URL доступен без отдельного sless_trigger.
|
||
|
||
package controllers
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"fmt"
|
||
"io"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
appsv1 "k8s.io/api/apps/v1"
|
||
corev1 "k8s.io/api/core/v1"
|
||
netv1 "k8s.io/api/networking/v1"
|
||
"k8s.io/apimachinery/pkg/api/errors"
|
||
"k8s.io/apimachinery/pkg/api/resource"
|
||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||
"k8s.io/apimachinery/pkg/runtime"
|
||
"k8s.io/client-go/kubernetes"
|
||
ctrl "sigs.k8s.io/controller-runtime"
|
||
"sigs.k8s.io/controller-runtime/pkg/client"
|
||
"sigs.k8s.io/controller-runtime/pkg/log"
|
||
|
||
slessv1alpha1 "gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/api/v1alpha1"
|
||
"gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/internal/builder"
|
||
"gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/internal/harbor"
|
||
)
|
||
|
||
// ServiceReconciler reconciles a Service object
|
||
type ServiceReconciler struct {
|
||
client.Client
|
||
Scheme *runtime.Scheme
|
||
Builder *builder.Builder
|
||
KubeClient kubernetes.Interface // typed client для чтения логов build-подов
|
||
RegistrySecret string // имя Secret с docker credentials
|
||
OperatorNamespace string // откуда копируем RegistrySecret в sless-fn-*
|
||
HarborClient *harbor.Client // nil — EnsureProject пропускается
|
||
ExternalURL string // базовый URL для Status.URL: {ExternalURL}/fn/{ns}/{name}
|
||
IngressHost string // fallback домен если ExternalURL не задан
|
||
}
|
||
|
||
//+kubebuilder:rbac:groups=sless.kube5s.ru,resources=services,verbs=get;list;watch;create;update;patch;delete
|
||
//+kubebuilder:rbac:groups=sless.kube5s.ru,resources=services/status,verbs=get;update;patch
|
||
//+kubebuilder:rbac:groups=sless.kube5s.ru,resources=services/finalizers,verbs=update
|
||
//+kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete
|
||
//+kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch;create;update;patch;delete
|
||
//+kubebuilder:rbac:groups=networking.k8s.io,resources=ingresses,verbs=get;list;watch;create;update;patch;delete
|
||
|
||
const serviceFinalizerName = "sless.kube5s.ru/service-finalizer"
|
||
|
||
// Reconcile — главный цикл управления Service.
|
||
// Pending → Building (запуск kaniko) → Ready (Deployment+Service+Ingress созданы, URL в Status)
|
||
func (r *ServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
|
||
logger := log.FromContext(ctx)
|
||
|
||
svc := &slessv1alpha1.Service{}
|
||
if err := r.Get(ctx, req.NamespacedName, svc); err != nil {
|
||
if errors.IsNotFound(err) {
|
||
return ctrl.Result{}, nil
|
||
}
|
||
return ctrl.Result{}, fmt.Errorf("get service: %w", err)
|
||
}
|
||
|
||
if !svc.DeletionTimestamp.IsZero() {
|
||
return r.handleServiceDeletion(ctx, svc)
|
||
}
|
||
|
||
if !containsString(svc.Finalizers, serviceFinalizerName) {
|
||
svc.Finalizers = append(svc.Finalizers, serviceFinalizerName)
|
||
if err := r.Update(ctx, svc); err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("add service finalizer: %w", err)
|
||
}
|
||
return ctrl.Result{Requeue: true}, nil
|
||
}
|
||
|
||
// Идентичная логике FunctionReconciler: если s3Key изменился — пересобираем образ.
|
||
builtKey := svc.Annotations["sless.kube5s.ru/last-built-s3key"]
|
||
needsBuild := svc.Spec.S3Key != "" && builtKey != svc.Spec.S3Key
|
||
|
||
if needsBuild && svc.Status.Phase != slessv1alpha1.ServicePhaseBuilding {
|
||
logger.Info("starting service build", "service", svc.Name)
|
||
return r.startServiceBuild(ctx, svc)
|
||
}
|
||
|
||
switch svc.Status.Phase {
|
||
case slessv1alpha1.ServicePhaseBuilding:
|
||
return r.checkServiceBuild(ctx, svc)
|
||
case slessv1alpha1.ServicePhaseReady:
|
||
return r.ensureServiceDeployment(ctx, svc)
|
||
}
|
||
|
||
return ctrl.Result{}, nil
|
||
}
|
||
|
||
// startServiceBuild проверяет наличие образа в registry и либо пропускает сборку,
|
||
// либо запускает kaniko Job. Идемпотентность: если код не менялся (тег = hash s3Key),
|
||
// образ уже в registry → deploy без пересборки.
|
||
func (r *ServiceReconciler) startServiceBuild(ctx context.Context, svc *slessv1alpha1.Service) (ctrl.Result, error) {
|
||
imageRef := r.Builder.ImageRef(svc.Namespace, svc.Name, svc.Spec.S3Key)
|
||
|
||
// Проверяем: образ с этим тегом уже существует в registry?
|
||
// Если да — пропускаем kaniko, сразу переходим в Ready с известным imageRef.
|
||
// Если registry недоступен — requeue, не запускаем сборку (kaniko тоже упадёт).
|
||
exists, err := r.Builder.ImageExists(ctx, imageRef)
|
||
if err != nil {
|
||
return ctrl.Result{RequeueAfter: 10 * time.Second}, fmt.Errorf("check image exists: %w", err)
|
||
}
|
||
if exists {
|
||
logger := log.FromContext(ctx)
|
||
logger.Info("image already exists in registry, skipping build", "imageRef", imageRef)
|
||
|
||
if svc.Annotations == nil {
|
||
svc.Annotations = map[string]string{}
|
||
}
|
||
svc.Annotations["sless.kube5s.ru/last-built-s3key"] = svc.Spec.S3Key
|
||
if err := r.Update(ctx, svc); err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("update service annotations (cache hit): %w", err)
|
||
}
|
||
|
||
svc.Status.Phase = slessv1alpha1.ServicePhaseReady
|
||
svc.Status.ImageRef = imageRef
|
||
svc.Status.Message = "Image restored from registry cache"
|
||
if err := r.Status().Update(ctx, svc); err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("update service status (cache hit): %w", err)
|
||
}
|
||
return ctrl.Result{}, nil
|
||
}
|
||
|
||
jobName, err := r.Builder.Build(ctx, svc.Namespace, svc.Name, svc.Spec.S3Key)
|
||
if err != nil {
|
||
return r.setServiceFailed(ctx, svc, fmt.Sprintf("failed to start build: %v", err))
|
||
}
|
||
|
||
if svc.Annotations == nil {
|
||
svc.Annotations = map[string]string{}
|
||
}
|
||
svc.Annotations["sless.kube5s.ru/build-job"] = jobName
|
||
svc.Annotations["sless.kube5s.ru/last-built-s3key"] = svc.Spec.S3Key
|
||
if err := r.Update(ctx, svc); err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("update service build annotations: %w", err)
|
||
}
|
||
|
||
svc.Status.Phase = slessv1alpha1.ServicePhaseBuilding
|
||
svc.Status.Message = "Building image: " + jobName
|
||
if err := r.Status().Update(ctx, svc); err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("update service status to building: %w", err)
|
||
}
|
||
|
||
return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
|
||
}
|
||
|
||
// checkServiceBuild проверяет статус kaniko Job.
|
||
func (r *ServiceReconciler) checkServiceBuild(ctx context.Context, svc *slessv1alpha1.Service) (ctrl.Result, error) {
|
||
jobName := svc.Annotations["sless.kube5s.ru/build-job"]
|
||
if jobName == "" {
|
||
svc.Status.Phase = slessv1alpha1.ServicePhasePending
|
||
_ = r.Status().Update(ctx, svc)
|
||
return ctrl.Result{Requeue: true}, nil
|
||
}
|
||
|
||
status, err := r.Builder.JobStatus(ctx, jobName)
|
||
if err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("check service build job: %w", err)
|
||
}
|
||
|
||
switch status {
|
||
case "running":
|
||
return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
|
||
case "succeeded":
|
||
imageRef := r.Builder.ImageRef(svc.Namespace, svc.Name, svc.Spec.S3Key)
|
||
svc.Status.Phase = slessv1alpha1.ServicePhaseReady
|
||
svc.Status.ImageRef = imageRef
|
||
svc.Status.Message = ""
|
||
now := metav1.Now()
|
||
svc.Status.LastBuiltAt = &now
|
||
if err := r.Status().Update(ctx, svc); err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("update service status to ready: %w", err)
|
||
}
|
||
_ = r.Builder.Cleanup(ctx, jobName)
|
||
return ctrl.Result{Requeue: true}, nil
|
||
case "failed":
|
||
logs := getServiceBuildPodLogs(ctx, r.KubeClient, r.OperatorNamespace, jobName)
|
||
msg := "build job failed"
|
||
if logs != "" {
|
||
msg = "build job failed:\n" + logs
|
||
}
|
||
return r.setServiceFailed(ctx, svc, msg)
|
||
}
|
||
|
||
return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
|
||
}
|
||
|
||
// ensureServiceDeployment создаёт или обновляет Deployment + k8s Service + Ingress.
|
||
// URL записывается в svc.Status.URL — доступен сразу без отдельного sless_trigger.
|
||
func (r *ServiceReconciler) ensureServiceDeployment(ctx context.Context, svc *slessv1alpha1.Service) (ctrl.Result, error) {
|
||
deployNS := "sless-fn-" + svc.Namespace
|
||
|
||
// Создаём namespace для функций/сервисов если не существует
|
||
ns := &corev1.Namespace{}
|
||
if err := r.Get(ctx, client.ObjectKey{Name: deployNS}, ns); err != nil {
|
||
if errors.IsNotFound(err) {
|
||
ns = &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: deployNS}}
|
||
if err := r.Create(ctx, ns); err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("create service namespace: %w", err)
|
||
}
|
||
if r.HarborClient != nil {
|
||
if err := r.HarborClient.EnsureProject(ctx, svc.Namespace); err != nil {
|
||
log.FromContext(ctx).Error(err, "harbor ensure project", "project", svc.Namespace)
|
||
}
|
||
}
|
||
} else {
|
||
return ctrl.Result{}, fmt.Errorf("get service namespace: %w", err)
|
||
}
|
||
}
|
||
|
||
// Копируем pull-секрет чтобы kubelet мог скачать образ из приватного registry
|
||
if r.RegistrySecret != "" && r.OperatorNamespace != "" {
|
||
if err := r.ensureServiceRegistrySecret(ctx, deployNS); err != nil {
|
||
log.FromContext(ctx).Error(err, "failed to ensure registry secret", "ns", deployNS)
|
||
}
|
||
}
|
||
|
||
// Deployment
|
||
desired := r.buildServiceDeployment(svc, deployNS)
|
||
existing := &appsv1.Deployment{}
|
||
err := r.Get(ctx, client.ObjectKey{Name: svc.Name, Namespace: deployNS}, existing)
|
||
if errors.IsNotFound(err) {
|
||
if err := r.Create(ctx, desired); err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("create service deployment: %w", err)
|
||
}
|
||
} else if err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("get service deployment: %w", err)
|
||
} else {
|
||
existing.Spec.Template.Spec.Containers[0].Image = svc.Status.ImageRef
|
||
existing.Spec.Template.Spec.Containers[0].Env = desired.Spec.Template.Spec.Containers[0].Env
|
||
// Обновляем ресурсы — иначе memory_mb из PUT не применяется к Deployment
|
||
existing.Spec.Template.Spec.Containers[0].Resources = desired.Spec.Template.Spec.Containers[0].Resources
|
||
existing.Spec.Template.Spec.ImagePullSecrets = desired.Spec.Template.Spec.ImagePullSecrets
|
||
if err := r.Update(ctx, existing); err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("update service deployment: %w", err)
|
||
}
|
||
}
|
||
|
||
// k8s Service — направляет трафик к Deployment
|
||
wantSvc := &corev1.Service{
|
||
ObjectMeta: metav1.ObjectMeta{
|
||
Name: svc.Name,
|
||
Namespace: deployNS,
|
||
Labels: map[string]string{"managed-by": "sless"},
|
||
},
|
||
Spec: corev1.ServiceSpec{
|
||
Selector: map[string]string{"app": svc.Name},
|
||
Ports: []corev1.ServicePort{{Port: 8080, Protocol: corev1.ProtocolTCP}},
|
||
},
|
||
}
|
||
existingSvc := &corev1.Service{}
|
||
if err := r.Get(ctx, client.ObjectKey{Name: svc.Name, Namespace: deployNS}, existingSvc); err != nil {
|
||
if errors.IsNotFound(err) {
|
||
if err := r.Create(ctx, wantSvc); err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("create k8s service: %w", err)
|
||
}
|
||
} else {
|
||
return ctrl.Result{}, fmt.Errorf("get k8s service: %w", err)
|
||
}
|
||
}
|
||
|
||
// URL и Ingress формируются либо через ExternalURL (прокси через API), либо через Ingress
|
||
var funcURL string
|
||
if r.ExternalURL != "" {
|
||
// ExternalURL/fn/{ns}/{name} — работает через sless-api прокси, без wildcard DNS
|
||
funcURL = fmt.Sprintf("%s/fn/%s/%s", r.ExternalURL, svc.Namespace, svc.Name)
|
||
} else {
|
||
// Fallback: Ingress с поддоменом (требует wildcard DNS *.IngressHost)
|
||
host := fmt.Sprintf("%s-%s.%s", svc.Name, svc.Namespace, r.IngressHost)
|
||
pathType := netv1.PathTypePrefix
|
||
wantIng := &netv1.Ingress{
|
||
ObjectMeta: metav1.ObjectMeta{
|
||
Name: svc.Name,
|
||
Namespace: deployNS,
|
||
Annotations: map[string]string{
|
||
"kubernetes.io/ingress.class": "nginx",
|
||
},
|
||
},
|
||
Spec: netv1.IngressSpec{
|
||
Rules: []netv1.IngressRule{{
|
||
Host: host,
|
||
IngressRuleValue: netv1.IngressRuleValue{
|
||
HTTP: &netv1.HTTPIngressRuleValue{
|
||
Paths: []netv1.HTTPIngressPath{{
|
||
Path: "/",
|
||
PathType: &pathType,
|
||
Backend: netv1.IngressBackend{
|
||
Service: &netv1.IngressServiceBackend{
|
||
Name: svc.Name,
|
||
Port: netv1.ServiceBackendPort{Number: 8080},
|
||
},
|
||
},
|
||
}},
|
||
},
|
||
},
|
||
}},
|
||
},
|
||
}
|
||
existingIng := &netv1.Ingress{}
|
||
if err := r.Get(ctx, client.ObjectKey{Name: svc.Name, Namespace: deployNS}, existingIng); err != nil {
|
||
if errors.IsNotFound(err) {
|
||
if err := r.Create(ctx, wantIng); err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("create service ingress: %w", err)
|
||
}
|
||
} else {
|
||
return ctrl.Result{}, fmt.Errorf("get service ingress: %w", err)
|
||
}
|
||
}
|
||
funcURL = "https://" + host
|
||
}
|
||
|
||
// Записываем URL в Status если изменился
|
||
if svc.Status.URL != funcURL {
|
||
svc.Status.URL = funcURL
|
||
if err := r.Status().Update(ctx, svc); err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("update service status url: %w", err)
|
||
}
|
||
}
|
||
|
||
// Периодический requeue — self-healing: если Deployment/Service/Ingress удалены вручную, контроллер их пересоздаст
|
||
return ctrl.Result{RequeueAfter: 60 * time.Second}, nil
|
||
}
|
||
|
||
// buildServiceDeployment формирует Deployment манифест.
|
||
func (r *ServiceReconciler) buildServiceDeployment(svc *slessv1alpha1.Service, namespace string) *appsv1.Deployment {
|
||
replicas := int32(1)
|
||
var envVars []corev1.EnvVar
|
||
|
||
// Сортируем ключи для стабильного порядка — нестабильный порядок env vars вызывает лишние rollout'ы
|
||
keys := make([]string, 0, len(svc.Spec.Env))
|
||
for k := range svc.Spec.Env {
|
||
keys = append(keys, k)
|
||
}
|
||
sort.Strings(keys)
|
||
for _, k := range keys {
|
||
envVars = append(envVars, corev1.EnvVar{Name: k, Value: svc.Spec.Env[k]})
|
||
}
|
||
// SLESS_ENTRYPOINT обязан быть передан в pod — runtime server использует его для загрузки функции.
|
||
// Если не передать, server.py/nodejs/go server упадёт на fallback handler.handle и не заметит
|
||
// неверный entrypoint, что маскирует проблему конфигурации.
|
||
if svc.Spec.Entrypoint != "" {
|
||
envVars = append(envVars, corev1.EnvVar{Name: "SLESS_ENTRYPOINT", Value: svc.Spec.Entrypoint})
|
||
}
|
||
|
||
return &appsv1.Deployment{
|
||
ObjectMeta: metav1.ObjectMeta{
|
||
Name: svc.Name,
|
||
Namespace: namespace,
|
||
Labels: map[string]string{"app": svc.Name, "managed-by": "sless"},
|
||
},
|
||
Spec: appsv1.DeploymentSpec{
|
||
Replicas: &replicas,
|
||
Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": svc.Name}},
|
||
Template: corev1.PodTemplateSpec{
|
||
ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": svc.Name}},
|
||
Spec: corev1.PodSpec{
|
||
Containers: []corev1.Container{{
|
||
Name: svc.Name,
|
||
Image: svc.Status.ImageRef,
|
||
Env: envVars,
|
||
Resources: corev1.ResourceRequirements{
|
||
Limits: corev1.ResourceList{
|
||
corev1.ResourceMemory: resource.MustParse(fmt.Sprintf("%dMi", svc.Spec.MemoryMB)),
|
||
},
|
||
},
|
||
}},
|
||
ImagePullSecrets: func() []corev1.LocalObjectReference {
|
||
if r.RegistrySecret != "" {
|
||
return []corev1.LocalObjectReference{{Name: r.RegistrySecret}}
|
||
}
|
||
return nil
|
||
}(),
|
||
},
|
||
},
|
||
},
|
||
}
|
||
}
|
||
|
||
// ensureServiceRegistrySecret копирует pull-секрет в namespace сервисов.
|
||
func (r *ServiceReconciler) ensureServiceRegistrySecret(ctx context.Context, targetNS string) error {
|
||
existing := &corev1.Secret{}
|
||
if err := r.Get(ctx, client.ObjectKey{Name: r.RegistrySecret, Namespace: targetNS}, existing); err == nil {
|
||
return nil
|
||
} else if !errors.IsNotFound(err) {
|
||
return fmt.Errorf("check secret: %w", err)
|
||
}
|
||
src := &corev1.Secret{}
|
||
if err := r.Get(ctx, client.ObjectKey{Name: r.RegistrySecret, Namespace: r.OperatorNamespace}, src); err != nil {
|
||
return fmt.Errorf("get source secret from %s: %w", r.OperatorNamespace, err)
|
||
}
|
||
copy := &corev1.Secret{
|
||
ObjectMeta: metav1.ObjectMeta{
|
||
Name: r.RegistrySecret,
|
||
Namespace: targetNS,
|
||
},
|
||
Type: src.Type,
|
||
Data: src.Data,
|
||
}
|
||
if err := r.Create(ctx, copy); err != nil {
|
||
if !errors.IsAlreadyExists(err) {
|
||
return fmt.Errorf("create secret in %s: %w", targetNS, err)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// handleServiceDeletion удаляет Deployment, k8s Service, Ingress и убирает finalizer.
|
||
func (r *ServiceReconciler) handleServiceDeletion(ctx context.Context, svc *slessv1alpha1.Service) (ctrl.Result, error) {
|
||
deployNS := "sless-fn-" + svc.Namespace
|
||
|
||
dep := &appsv1.Deployment{}
|
||
if err := r.Get(ctx, client.ObjectKey{Name: svc.Name, Namespace: deployNS}, dep); err == nil {
|
||
_ = r.Delete(ctx, dep)
|
||
}
|
||
|
||
// Убиваем kaniko Job если сборка шла в момент удаления
|
||
if jobName := svc.Annotations["sless.kube5s.ru/build-job"]; jobName != "" {
|
||
_ = r.Builder.Cleanup(ctx, jobName)
|
||
}
|
||
|
||
k8sSvc := &corev1.Service{}
|
||
if err := r.Get(ctx, client.ObjectKey{Name: svc.Name, Namespace: deployNS}, k8sSvc); err == nil {
|
||
_ = r.Delete(ctx, k8sSvc)
|
||
}
|
||
|
||
ing := &netv1.Ingress{}
|
||
if err := r.Get(ctx, client.ObjectKey{Name: svc.Name, Namespace: deployNS}, ing); err == nil {
|
||
_ = r.Delete(ctx, ing)
|
||
}
|
||
|
||
svc.Finalizers = removeString(svc.Finalizers, serviceFinalizerName)
|
||
if err := r.Update(ctx, svc); err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("remove service finalizer: %w", err)
|
||
}
|
||
return ctrl.Result{}, nil
|
||
}
|
||
|
||
// setServiceFailed переводит сервис в фазу Failed.
|
||
func (r *ServiceReconciler) setServiceFailed(ctx context.Context, svc *slessv1alpha1.Service, msg string) (ctrl.Result, error) {
|
||
svc.Status.Phase = slessv1alpha1.ServicePhaseFailed
|
||
svc.Status.Message = msg
|
||
if err := r.Status().Update(ctx, svc); err != nil {
|
||
return ctrl.Result{}, fmt.Errorf("update service status to failed: %w", err)
|
||
}
|
||
return ctrl.Result{}, nil
|
||
}
|
||
|
||
// getServiceBuildPodLogs читает логи kaniko пода (последние 50 строк).
|
||
// Идентична getBuildPodLogs из function_controller.go, но вынесена в service_controller
|
||
// чтобы не создавать shared-помощника ради двух вызовов.
|
||
func getServiceBuildPodLogs(ctx context.Context, kube kubernetes.Interface, namespace, jobName string) string {
|
||
if kube == nil {
|
||
return ""
|
||
}
|
||
pods, err := kube.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{
|
||
LabelSelector: "job-name=" + jobName,
|
||
})
|
||
if err != nil || len(pods.Items) == 0 {
|
||
return ""
|
||
}
|
||
req := kube.CoreV1().Pods(namespace).GetLogs(pods.Items[0].Name, &corev1.PodLogOptions{})
|
||
stream, err := req.Stream(ctx)
|
||
if err != nil {
|
||
return ""
|
||
}
|
||
defer stream.Close()
|
||
buf := new(bytes.Buffer)
|
||
_, _ = io.Copy(buf, stream)
|
||
raw := strings.TrimSpace(buf.String())
|
||
if raw == "" {
|
||
return ""
|
||
}
|
||
lines := strings.Split(raw, "\n")
|
||
if len(lines) > 50 {
|
||
lines = lines[len(lines)-50:]
|
||
}
|
||
return strings.Join(lines, "\n")
|
||
}
|
||
|
||
// SetupWithManager sets up the controller with the Manager.
|
||
func (r *ServiceReconciler) SetupWithManager(mgr ctrl.Manager) error {
|
||
return ctrl.NewControllerManagedBy(mgr).
|
||
For(&slessv1alpha1.Service{}).
|
||
Complete(r)
|
||
}
|