devstar-create-from-template/modules/k8s/application/application.go

package application

import (
	"context"
	"encoding/json"
	"fmt"

	applicationv1 "code.gitea.io/gitea/modules/k8s/api/application/v1"
	"code.gitea.io/gitea/modules/log"

	application_errors "code.gitea.io/gitea/modules/k8s/application/errors"
	apimachinery_apis_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	apimachinery_apis_v1_unstructured "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	apimachinery_runtime_utils "k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	apimachinery_watch "k8s.io/apimachinery/pkg/watch"
	dynamic_client "k8s.io/client-go/dynamic"
)

// Application 资源定义
var applicationGroupVersionResource = schema.GroupVersionResource{
	Group:    "application.devstar.cn",
	Version:  "v1",
	Resource: "applications",
}

// IsApplicationStatusReady 判断 Application 是否就绪
func IsApplicationStatusReady(applicationStatus *applicationv1.ApplicationStatus) bool {
	if applicationStatus == nil {
		return false
	}

	// 如果不是 Running 状态，则未就绪
	if applicationStatus.Phase != "Running" {
		return false
	}

	// 零副本应用被认为是就绪的（暂停状态）
	if applicationStatus.Replicas == 0 && applicationStatus.ReadyReplicas == 0 {
		return true
	}

	// 有副本的应用需要至少有一个就绪副本
	return applicationStatus.ReadyReplicas > 0
}

// GetApplication 获取 Application
func GetApplication(ctx context.Context, client dynamic_client.Interface, opts *GetApplicationOptions) (*applicationv1.Application, error) {
	// 参数检查
	if ctx == nil || opts == nil || len(opts.Namespace) == 0 || len(opts.Name) == 0 {
		return nil, application_errors.ErrIllegalApplicationParameters{
			FieldList: []string{"ctx", "opts", "opts.Name", "opts.Namespace"},
			Message:   "cannot be nil",
		}
	}

	// 获取 Application 资源
	applicationUnstructured, err := client.Resource(applicationGroupVersionResource).
		Namespace(opts.Namespace).Get(ctx, opts.Name, opts.GetOptions)
	if err != nil {
		return nil, application_errors.ErrOperateApplication{
			Action:  "Get Application through k8s API Server",
			Message: err.Error(),
		}
	}

	// 转换为 Application 对象
	application := &applicationv1.Application{}
	err = apimachinery_runtime_utils.DefaultUnstructuredConverter.
		FromUnstructured(applicationUnstructured.Object, application)
	if err != nil {
		return nil, application_errors.ErrOperateApplication{
			Action:  "Convert k8s API Server unstructured response into Application",
			Message: err.Error(),
		}
	}

	// 检查是否需要等待就绪
	if !IsApplicationStatusReady(&application.Status) && opts.Wait {
		applicationStatusVO, err := waitUntilApplicationReadyWithTimeout(ctx, client, opts)
		if err != nil {
			return nil, application_errors.ErrOperateApplication{
				Action:  "wait for k8s Application to be ready",
				Message: err.Error(),
			}
		}
		application.Status = *applicationStatusVO
	}

	return application, nil
}

// CreateApplication 创建 Application
func CreateApplication(ctx context.Context, client dynamic_client.Interface, opts *CreateApplicationOptions) (*applicationv1.Application, error) {
	log.Info("Creating Application with options: name=%s, namespace=%s, image=%s",
		opts.Name, opts.Namespace, opts.Template.Image)

	// 处理默认的 Resources
	var resources applicationv1.ResourceRequirements
	if opts.Resources != nil {
		resources = *opts.Resources
	}
	// 如果 Resources 为 nil，使用空的 ResourceRequirements（所有字段都是零值）

	// 创建 Application 资源定义
	application := &applicationv1.Application{
		TypeMeta: apimachinery_apis_v1.TypeMeta{
			Kind:       "Application",
			APIVersion: "application.devstar.cn/v1",
		},
		ObjectMeta: apimachinery_apis_v1.ObjectMeta{
			Name:      opts.Name,
			Namespace: opts.Namespace,
			Labels: map[string]string{
				"app.kubernetes.io/name":       opts.Name,
				"app.kubernetes.io/component":  opts.Component,
				"app.kubernetes.io/managed-by": "devstar",
			},
		},
		Spec: applicationv1.ApplicationSpec{
			Template:      opts.Template,
			Replicas:      opts.Replicas,
			Environment:   opts.Environment,
			Resources:     resources,
			Expose:        opts.Expose,
			Service:       opts.Service,
			NetworkPolicy: opts.NetworkPolicy,
			TrafficPolicy: opts.TrafficPolicy,
		},
	}

	// 转换为 JSON
	jsonData, err := json.Marshal(application)
	if err != nil {
		log.Error("Failed to marshal Application to JSON: %v", err)
		return nil, application_errors.ErrOperateApplication{
			Action:  "Marshal JSON",
			Message: err.Error(),
		}
	}

	log.Debug("Generated JSON for Application:\n%s", string(jsonData))

	// 转换为 Unstructured 对象
	unstructuredObj := &apimachinery_apis_v1_unstructured.Unstructured{}
	err = unstructuredObj.UnmarshalJSON(jsonData)
	if err != nil {
		log.Error("Failed to unmarshal JSON to Unstructured: %v", err)
		return nil, application_errors.ErrOperateApplication{
			Action:  "Unmarshal JSON to Unstructured",
			Message: err.Error(),
		}
	}

	// 创建资源
	log.Info("Creating Application resource in namespace %s", opts.Namespace)
	result, err := client.Resource(applicationGroupVersionResource).
		Namespace(opts.Namespace).Create(ctx, unstructuredObj, opts.CreateOptions)
	if err != nil {
		log.Error("Failed to create Application: %v", err)
		return nil, application_errors.ErrOperateApplication{
			Action:  "create Application via Dynamic Client",
			Message: err.Error(),
		}
	}

	log.Info("Application resource created successfully")

	// 转换结果
	resultJSON, err := result.MarshalJSON()
	if err != nil {
		return nil, application_errors.ErrOperateApplication{
			Action:  "Marshal result JSON",
			Message: err.Error(),
		}
	}

	createdApplication := &applicationv1.Application{}
	if err := json.Unmarshal(resultJSON, createdApplication); err != nil {
		return nil, application_errors.ErrOperateApplication{
			Action:  "Unmarshal result JSON",
			Message: err.Error(),
		}
	}

	return createdApplication, nil
}

// UpdateApplication 更新 Application
func UpdateApplication(ctx context.Context, client dynamic_client.Interface, opts *UpdateApplicationOptions) (*applicationv1.Application, error) {
	log.Info("Updating Application: name=%s, namespace=%s", opts.Name, opts.Namespace)

	// 转换为 JSON
	jsonData, err := json.Marshal(opts.Application)
	if err != nil {
		return nil, application_errors.ErrOperateApplication{
			Action:  "Marshal Application to JSON",
			Message: err.Error(),
		}
	}

	// 转换为 Unstructured
	unstructuredObj := &apimachinery_apis_v1_unstructured.Unstructured{}
	err = unstructuredObj.UnmarshalJSON(jsonData)
	if err != nil {
		return nil, application_errors.ErrOperateApplication{
			Action:  "Unmarshal JSON to Unstructured",
			Message: err.Error(),
		}
	}

	// 更新资源
	result, err := client.Resource(applicationGroupVersionResource).
		Namespace(opts.Namespace).Update(ctx, unstructuredObj, opts.UpdateOptions)
	if err != nil {
		return nil, application_errors.ErrOperateApplication{
			Action:  "update Application via Dynamic Client",
			Message: err.Error(),
		}
	}

	// 转换结果
	resultJSON, err := result.MarshalJSON()
	if err != nil {
		return nil, application_errors.ErrOperateApplication{
			Action:  "Marshal result JSON",
			Message: err.Error(),
		}
	}

	updatedApplication := &applicationv1.Application{}
	if err := json.Unmarshal(resultJSON, updatedApplication); err != nil {
		return nil, application_errors.ErrOperateApplication{
			Action:  "Unmarshal result JSON",
			Message: err.Error(),
		}
	}

	return updatedApplication, nil
}

// DeleteApplication 删除 Application
func DeleteApplication(ctx context.Context, client dynamic_client.Interface, opts *DeleteApplicationOptions) error {
	if ctx == nil || opts == nil || len(opts.Namespace) == 0 || len(opts.Name) == 0 {
		return application_errors.ErrIllegalApplicationParameters{
			FieldList: []string{"ctx", "opts", "opts.Name", "opts.Namespace"},
			Message:   "cannot be nil",
		}
	}

	err := client.Resource(applicationGroupVersionResource).
		Namespace(opts.Namespace).Delete(ctx, opts.Name, opts.DeleteOptions)
	if err != nil {
		log.Warn("Failed to delete Application '%s' in namespace '%s': %s",
			opts.Name, opts.Namespace, err.Error())
		return application_errors.ErrOperateApplication{
			Action:  fmt.Sprintf("delete application '%s' in namespace '%s'", opts.Name, opts.Namespace),
			Message: err.Error(),
		}
	}
	return nil
}

// ListApplications 列出 Applications
func ListApplications(ctx context.Context, client dynamic_client.Interface, opts *ListApplicationsOptions) (*applicationv1.ApplicationList, error) {
	if ctx == nil || opts == nil || len(opts.Namespace) == 0 {
		return nil, application_errors.ErrIllegalApplicationParameters{
			FieldList: []string{"ctx", "namespace"},
			Message:   "cannot be empty",
		}
	}

	list, err := client.Resource(applicationGroupVersionResource).
		Namespace(opts.Namespace).List(ctx, opts.ListOptions)
	if err != nil {
		return nil, application_errors.ErrOperateApplication{
			Action:  fmt.Sprintf("List Application in namespace '%s'", opts.Namespace),
			Message: err.Error(),
		}
	}

	// 转换为 ApplicationList
	jsonData, err := list.MarshalJSON()
	if err != nil {
		return nil, application_errors.ErrOperateApplication{
			Action:  "verify JSON data of Application List",
			Message: err.Error(),
		}
	}

	applicationList := &applicationv1.ApplicationList{}
	if err := json.Unmarshal(jsonData, applicationList); err != nil {
		return nil, application_errors.ErrOperateApplication{
			Action:  "deserialize Application List data",
			Message: err.Error(),
		}
	}

	return applicationList, nil
}

// 等待 Application 就绪的辅助函数
func waitUntilApplicationReadyWithTimeout(ctx context.Context, client dynamic_client.Interface, opts *GetApplicationOptions) (*applicationv1.ApplicationStatus, error) {
	// 监听 Application 状态变化，等待就绪
	watcherTimeoutSeconds := int64(300) // 5分钟超时
	watcher, err := client.Resource(applicationGroupVersionResource).
		Namespace(opts.Namespace).Watch(ctx, apimachinery_apis_v1.ListOptions{
		FieldSelector:  fmt.Sprintf("metadata.name=%s", opts.Name),
		Watch:          true,
		TimeoutSeconds: &watcherTimeoutSeconds,
	})
	if err != nil {
		return nil, application_errors.ErrOperateApplication{
			Action:  "register watcher of Application Readiness",
			Message: err.Error(),
		}
	}
	defer watcher.Stop()

	for event := range watcher.ResultChan() {
		switch event.Type {
		case apimachinery_watch.Added, apimachinery_watch.Modified:
			if applicationUnstructured, ok := event.Object.(*apimachinery_apis_v1_unstructured.Unstructured); ok {
				application := &applicationv1.Application{}
				err = apimachinery_runtime_utils.DefaultUnstructuredConverter.
					FromUnstructured(applicationUnstructured.Object, application)
				if err == nil && IsApplicationStatusReady(&application.Status) {
					return &application.Status, nil
				}
			}
		case apimachinery_watch.Error:
			apimachineryApiMetav1Status, ok := event.Object.(*apimachinery_apis_v1.Status)
			if ok {
				return nil, application_errors.ErrOperateApplication{
					Action: fmt.Sprintf("wait for Application '%s' in namespace '%s' to be ready",
						opts.Name, opts.Namespace),
					Message: fmt.Sprintf("Watcher error: %v", apimachineryApiMetav1Status.Message),
				}
			}
		case apimachinery_watch.Deleted:
			return nil, application_errors.ErrOperateApplication{
				Action:  fmt.Sprintf("wait for Application '%s' in namespace '%s'", opts.Name, opts.Namespace),
				Message: fmt.Sprintf("Application '%s' has been deleted", opts.Name),
			}
		}
	}

	return nil, application_errors.ErrOperateApplication{
		Action:  "wait for Application to be ready",
		Message: "timeout waiting for Application to be ready",
	}
}