devstar/modules/devstar_devcontainer/k8s_agent/CreateDevcontainer.go

package k8s_agent

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

	devcontainer_api_v1 "code.gitea.io/gitea/modules/devstar_devcontainer/k8s_agent/api/v1"
	devcontainer_dto "code.gitea.io/gitea/modules/devstar_devcontainer/k8s_agent/dto"
	devcontainer_k8s_agent_modules_errors "code.gitea.io/gitea/modules/devstar_devcontainer/k8s_agent/errors"
	apimachinery_apis_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	apimachinery_apis_v1_unstructured "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	apimachinery_watch "k8s.io/apimachinery/pkg/watch"
	dynamic_client "k8s.io/client-go/dynamic"
)

// CreateDevcontainer 创建开发容器
func CreateDevcontainer(ctx *context.Context, client dynamic_client.Interface, opts *devcontainer_dto.CreateDevcontainerOptions) (*devcontainer_api_v1.DevcontainerApp, error) {

	if ctx == nil || opts == nil {
		return nil, devcontainer_k8s_agent_modules_errors.ErrIllegalDevcontainerParameters{
			FieldList: []string{"ctx", "opts"},
			Message:   "cannot be nil",
		}
	}

	devcontainerApp := &devcontainer_api_v1.DevcontainerApp{
		TypeMeta: apimachinery_apis_v1.TypeMeta{
			Kind:       "DevcontainerApp",
			APIVersion: "devcontainer.devstar.cn/v1",
		},
		ObjectMeta: apimachinery_apis_v1.ObjectMeta{
			Name:      opts.Name,
			Namespace: opts.Namespace,
		},
		Spec: devcontainer_api_v1.DevcontainerAppSpec{
			StatefulSet: devcontainer_api_v1.StatefulSetSpec{
				Image:         opts.Image,
				Command:       opts.CommandList,
				ContainerPort: opts.ContainerPort,
			},
		},
	}

	jsonData, err := json.Marshal(devcontainerApp)
	if err != nil {
		return nil, devcontainer_k8s_agent_modules_errors.ErrOperateDevcontainer{
			Action:  "Marshal JSON",
			Message: err.Error(),
		}
	}

	unstructuredObj := &apimachinery_apis_v1_unstructured.Unstructured{}
	_, _, err = apimachinery_apis_v1_unstructured.UnstructuredJSONScheme.Decode(jsonData, &groupVersionKind, unstructuredObj)
	if err != nil {
		return nil, devcontainer_k8s_agent_modules_errors.ErrOperateDevcontainer{
			Action:  "build unstructured obj",
			Message: err.Error(),
		}
	}

	// 创建 DevContainer Status.nodePortAssigned 信息
	_, err = client.Resource(groupVersionResource).Namespace(opts.Namespace).Create(*ctx, unstructuredObj, opts.CreateOptions)
	if err != nil {
		return nil, devcontainer_k8s_agent_modules_errors.ErrOperateDevcontainer{
			Action:  "create DevContainer via Dynamic Client",
			Message: err.Error(),
		}
	}

	// 注册 watcher 监听 DevContainer Status.nodePortAssigned 信息
	watcherTimeoutSeconds := int64(3)
	watcher, err := client.Resource(groupVersionResource).Namespace(opts.Namespace).Watch(*ctx, apimachinery_apis_v1.ListOptions{
		FieldSelector: fmt.Sprintf("metadata.name=%s", opts.Name),
		//Watch:               false,
		TimeoutSeconds: &watcherTimeoutSeconds,
		Limit:          1,
	})
	if err != nil {
		return nil, devcontainer_k8s_agent_modules_errors.ErrOperateDevcontainer{
			Action:  "register watcher of DevContainer NodePort",
			Message: err.Error(),
		}
	}
	defer watcher.Stop()

	//log.Info("     =====   开始监听 DevContainer NodePort 分配信息  =====")
	var nodePortAssigned int64
	for event := range watcher.ResultChan() {
		switch event.Type {
		case apimachinery_watch.Modified:
			if devcontainerUnstructured, ok := event.Object.(*apimachinery_apis_v1_unstructured.Unstructured); ok {

				statusDevcontainer, ok, err := apimachinery_apis_v1_unstructured.NestedMap(devcontainerUnstructured.Object, "status")
				if err == nil && ok {
					nodePortAssigned = statusDevcontainer["nodePortAssigned"].(int64)
					if 30000 <= nodePortAssigned && nodePortAssigned <= 32767 {
						devcontainerApp.Status.NodePortAssigned = uint16(nodePortAssigned)
						//log.Info("DevContainer NodePort Status 更新完成，最新 NodePort = %v", nodePortAssigned)
						//break
						// 收到 NodePort Service MODIFIED 消息后，更新 NodePort，直接返回，不再处理后续 Event (否则必须超时3秒才得到 NodePort)
						return devcontainerApp, nil
					}
				}
			}
		}
	}
	//log.Info("     =====   结束监听 DevContainer NodePort 分配信息  =====")

	/*
		//    目前需要更新的字段只有 devcontainerInCluster.Status.NodePortAssigned
		//    如果后续有其他较多的需要更新的字段，可以考虑重新查询，目前，暂时只考虑 直接更新内存缓存对象，然后返回即可
		//    避免对 k8s API Server 造成访问压力
		//response, err := client.Resource(groupVersionResource).Namespace(opts.Namespace).Get(*ctx, opts.Name, apimachinery_apis_v1.GetOptions{})
		//jsonData, err = response.MarshalJSON()
		//devcontainerInCluster := &devcontainer_api_v1.DevcontainerApp{}
		//err = json.Unmarshal(jsonData, devcontainerInCluster)
		//if err != nil {
		//	return nil, devcontainer_errors.ErrOperateDevcontainer{
		//		Action:  "parse response result of in-cluster DevContainer",
		//		Message: err.Error(),
		//	}
		//}
	*/
	return devcontainerApp, nil
}