Files
xinfra/server/internal/service/container_service.go
T

786 lines
22 KiB
Go

package service
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/url"
"strconv"
"strings"
"time"
"github.com/1024XEngineer/xinfra/server/internal/model"
)
type ContainerServiceSummary struct {
BusinessLineID uint64 `json:"businessLineId"`
BusinessLineName string `json:"businessLineName"`
Namespaces int `json:"namespaces"`
Clusters int `json:"clusters"`
Nodes int `json:"nodes"`
ReadyNodes int `json:"readyNodes"`
SchedulableNodes int `json:"schedulableNodes"`
Workloads int `json:"workloads"`
Pods int `json:"pods"`
ReadyPods int `json:"readyPods"`
CPUUsed float64 `json:"cpuUsed"`
CPUTotal float64 `json:"cpuTotal"`
CPUPercent int `json:"cpuPercent"`
MemoryUsed float64 `json:"memoryUsed"`
MemoryTotal float64 `json:"memoryTotal"`
MemoryPercent int `json:"memoryPercent"`
Errors []string `json:"errors,omitempty"`
}
type ContainerWorkload struct {
Name string `json:"name"`
Cluster string `json:"cluster"`
Namespace string `json:"namespace"`
Workload string `json:"workload"`
Ready string `json:"ready"`
ReadyPods int `json:"readyPods"`
Pods int `json:"pods"`
Image string `json:"image"`
BusinessLine string `json:"biz"`
Status string `json:"status"`
StatusClass string `json:"statusClass"`
}
type ContainerCluster struct {
Name string `json:"name"`
Zone string `json:"zone"`
ZoneClass string `json:"zoneClass"`
Status string `json:"status"`
StatusClass string `json:"statusClass"`
Nodes int `json:"nodes"`
ReadyNodes int `json:"readyNodes"`
CPU int `json:"cpu"`
CPUClass string `json:"cpuClass"`
Version string `json:"version"`
Calico string `json:"calico"`
}
type ContainerNode struct {
Cluster string `json:"cluster"`
Name string `json:"name"`
IP string `json:"ip"`
Label string `json:"label"`
Taint string `json:"taint"`
Spec string `json:"spec"`
CPU int `json:"cpu"`
CPUClass string `json:"cpuClass"`
Status string `json:"status"`
StatusClass string `json:"statusClass"`
}
type ContainerServiceData struct {
Summary ContainerServiceSummary `json:"summary"`
Workloads []ContainerWorkload `json:"items"`
Clusters []ContainerCluster `json:"clusters"`
Nodes []ContainerNode `json:"nodes"`
}
type ContainerBusinessLineQuery struct {
BusinessLineID uint64
BusinessLineName string
Namespaces []model.BusinessLineWayneNamespace
}
type WayneDeploymentHistory struct {
ID int64 `json:"id"`
Type int `json:"type"`
ResourceID int64 `json:"resourceId"`
ResourceName string `json:"resourceName"`
TemplateID int64 `json:"templateId"`
Cluster string `json:"cluster"`
Status int `json:"status"`
Message string `json:"message"`
User string `json:"user"`
CreateTime string `json:"createTime"`
CreatedAt time.Time `json:"-"`
BusinessLineID uint64 `json:"businessLineId"`
NamespaceID uint64 `json:"namespaceId"`
}
func (s *WayneRoleBindingService) ContainerServiceData(ctx context.Context, query ContainerBusinessLineQuery) (*ContainerServiceData, error) {
summary := ContainerServiceSummary{
BusinessLineID: query.BusinessLineID,
BusinessLineName: query.BusinessLineName,
Namespaces: len(query.Namespaces),
}
workloads := make([]ContainerWorkload, 0)
clusterNodes := map[string]wayneNodeData{}
seenWorkloads := map[string]struct{}{}
errors := make([]string, 0)
for _, binding := range query.Namespaces {
namespaceDetail, err := s.getWayneNamespace(ctx, binding.WayneNamespaceID)
if err != nil {
errors = append(errors, fmt.Sprintf("namespace %d: %v", binding.WayneNamespaceID, err))
continue
}
kubeNamespace := strings.TrimSpace(binding.KubeNamespace)
if kubeNamespace == "" {
kubeNamespace = namespaceDetail.KubeNamespace
}
clusters := namespaceDetail.Clusters()
if len(clusters) == 0 {
errors = append(errors, fmt.Sprintf("namespace %d has no clusterMeta", binding.WayneNamespaceID))
continue
}
apps, err := s.listWayneNamespaceApps(ctx, binding.WayneNamespaceID)
if err != nil {
errors = append(errors, fmt.Sprintf("namespace %d apps: %v", binding.WayneNamespaceID, err))
continue
}
if len(apps) == 0 {
errors = append(errors, fmt.Sprintf("namespace %d has no apps for deployment query", binding.WayneNamespaceID))
continue
}
appID := apps[0].ID
for _, cluster := range clusters {
if _, ok := clusterNodes[cluster]; !ok {
nodes, err := s.getWayneNodes(ctx, cluster)
if err != nil {
errors = append(errors, fmt.Sprintf("cluster %s nodes: %v", cluster, err))
} else {
clusterNodes[cluster] = nodes
}
}
items, err := s.listWayneDeployments(ctx, appID, kubeNamespace, cluster)
if err != nil {
errors = append(errors, fmt.Sprintf("namespace %d cluster %s app %d deployments: %v", binding.WayneNamespaceID, cluster, appID, err))
continue
}
for _, item := range items {
workload := containerWorkloadFromMap(item, cluster, kubeNamespace, query.BusinessLineName)
if workload.Name == "" {
continue
}
key := strings.Join([]string{workload.Cluster, workload.Namespace, workload.Workload, workload.Name}, "\x00")
if _, ok := seenWorkloads[key]; ok {
continue
}
seenWorkloads[key] = struct{}{}
workloads = append(workloads, workload)
summary.Pods += workload.Pods
summary.ReadyPods += workload.ReadyPods
}
}
}
clusters := make([]ContainerCluster, 0, len(clusterNodes))
allNodes := make([]ContainerNode, 0)
for clusterName, nodes := range clusterNodes {
summary.Nodes += nodes.Total
summary.ReadyNodes += nodes.Ready
summary.SchedulableNodes += nodes.Schedulable
summary.CPUUsed += nodes.CPUUsed
summary.CPUTotal += nodes.CPUTotal
summary.MemoryUsed += nodes.MemoryUsed
summary.MemoryTotal += nodes.MemoryTotal
clusters = append(clusters, containerClusterFromWayne(clusterName, nodes))
for _, node := range nodes.Nodes {
allNodes = append(allNodes, containerNodeFromWayne(clusterName, node, query.BusinessLineName))
}
}
summary.Clusters = len(clusterNodes)
summary.Workloads = len(workloads)
summary.CPUPercent = ratioPercent(summary.CPUUsed, summary.CPUTotal)
summary.MemoryPercent = ratioPercent(summary.MemoryUsed, summary.MemoryTotal)
summary.Errors = errors
sortContainerClusters(clusters)
sortContainerNodes(allNodes)
return &ContainerServiceData{Summary: summary, Workloads: workloads, Clusters: clusters, Nodes: allNodes}, nil
}
func (s *WayneRoleBindingService) ListDeploymentHistories(ctx context.Context, namespaces []model.BusinessLineWayneNamespace, limit int) ([]WayneDeploymentHistory, error) {
if limit <= 0 {
limit = 100
}
histories := make([]WayneDeploymentHistory, 0)
seenDeployments := map[int64]struct{}{}
for _, binding := range namespaces {
apps, err := s.listWayneNamespaceApps(ctx, binding.WayneNamespaceID)
if err != nil {
return nil, err
}
for _, app := range apps {
deployments, err := s.listWayneAppDeployments(ctx, app.ID)
if err != nil {
return nil, err
}
for _, deployment := range deployments {
if deployment.ID == 0 {
continue
}
if _, ok := seenDeployments[deployment.ID]; ok {
continue
}
seenDeployments[deployment.ID] = struct{}{}
items, err := s.listWaynePublishHistories(ctx, deployment.ID, 20)
if err != nil {
return nil, err
}
for _, item := range items {
item.BusinessLineID = binding.BusinessLineID
item.NamespaceID = binding.WayneNamespaceID
if item.ResourceName == "" {
item.ResourceName = deployment.Name
}
histories = append(histories, item)
}
}
}
}
sortWayneDeploymentHistories(histories)
if len(histories) > limit {
histories = histories[:limit]
}
return histories, nil
}
func (s *WayneRoleBindingService) GetDeploymentHistory(ctx context.Context, namespaces []model.BusinessLineWayneNamespace, resourceID, historyID int64) (*WayneDeploymentHistory, error) {
histories, err := s.ListDeploymentHistories(ctx, namespaces, 500)
if err != nil {
return nil, err
}
for _, history := range histories {
if history.ResourceID == resourceID && history.ID == historyID {
return &history, nil
}
}
return nil, fmt.Errorf("wayne deployment history %d was not found", historyID)
}
func (s *WayneRoleBindingService) getWayneNamespace(ctx context.Context, id uint64) (wayneNamespaceDetail, error) {
result, err := s.callRaw(ctx, http.MethodGet, fmt.Sprintf("/api/v1/namespaces/%d", id), nil)
if err != nil {
return wayneNamespaceDetail{}, err
}
return parseWayneNamespaceDetail(result.Body)
}
func (s *WayneRoleBindingService) listWayneNamespaceApps(ctx context.Context, namespaceID uint64) ([]wayneApp, error) {
values := url.Values{}
values.Set("pageNo", "1")
values.Set("pageSize", "500")
result, err := s.callRaw(ctx, http.MethodGet, fmt.Sprintf("/api/v1/namespaces/%d/apps?%s", namespaceID, values.Encode()), nil)
if err != nil {
return nil, err
}
return parseWayneApps(result.Body)
}
func (s *WayneRoleBindingService) listWayneAppDeployments(ctx context.Context, appID uint64) ([]wayneDeployment, error) {
values := url.Values{}
values.Set("pageNo", "1")
values.Set("pageSize", "500")
values.Set("deleted", "false")
result, err := s.callRaw(ctx, http.MethodGet, fmt.Sprintf("/api/v1/apps/%d/deployments?%s", appID, values.Encode()), nil)
if err != nil {
return nil, err
}
return parseWayneDeployments(result.Body)
}
func (s *WayneRoleBindingService) listWaynePublishHistories(ctx context.Context, resourceID int64, pageSize int) ([]WayneDeploymentHistory, error) {
values := url.Values{}
values.Set("pageNo", "1")
values.Set("pageSize", strconv.Itoa(pageSize))
values.Set("type", "0")
values.Set("resourceId", strconv.FormatInt(resourceID, 10))
values.Set("sortby", "-createTime")
result, err := s.callRaw(ctx, http.MethodGet, "/api/v1/publish/histories?"+values.Encode(), nil)
if err != nil {
return nil, err
}
return parseWaynePublishHistories(result.Body)
}
func (s *WayneRoleBindingService) getWayneNodes(ctx context.Context, cluster string) (wayneNodeData, error) {
result, err := s.callRaw(ctx, http.MethodGet, fmt.Sprintf("/api/v1/kubernetes/nodes/clusters/%s", url.PathEscape(cluster)), nil)
if err != nil {
return wayneNodeData{}, err
}
return parseWayneNodeSummary(result.Body)
}
func (s *WayneRoleBindingService) listWayneDeployments(ctx context.Context, appID uint64, namespace string, cluster string) ([]map[string]any, error) {
values := url.Values{}
values.Set("pageNo", "1")
values.Set("pageSize", "500")
path := fmt.Sprintf(
"/api/v1/kubernetes/apps/%d/deployments/namespaces/%s/clusters/%s?%s",
appID,
url.PathEscape(namespace),
url.PathEscape(cluster),
values.Encode(),
)
result, err := s.callRaw(ctx, http.MethodGet, path, nil)
if err != nil {
return nil, err
}
return parseWayneMapList(result.Body)
}
type wayneNamespaceDetail struct {
ID uint64 `json:"id"`
Name string `json:"name"`
KubeNamespace string `json:"kubeNamespace"`
MetaData string `json:"metaData"`
}
func (n wayneNamespaceDetail) Clusters() []string {
var meta struct {
ClusterMeta map[string]json.RawMessage `json:"clusterMeta"`
}
if err := json.Unmarshal([]byte(n.MetaData), &meta); err != nil {
return nil
}
clusters := make([]string, 0, len(meta.ClusterMeta))
for cluster := range meta.ClusterMeta {
if strings.TrimSpace(cluster) != "" {
clusters = append(clusters, cluster)
}
}
return clusters
}
type wayneApp struct {
ID uint64 `json:"id"`
Name string `json:"name"`
}
type wayneDeployment struct {
ID int64 `json:"id"`
Name string `json:"name"`
}
type wayneNodeData struct {
Total int
Ready int
Schedulable int
CPUUsed float64
CPUTotal float64
MemoryUsed float64
MemoryTotal float64
Nodes []wayneNode
}
type wayneNode struct {
Name string `json:"name"`
Labels map[string]string `json:"labels"`
Spec struct {
Unschedulable bool `json:"unschedulable"`
Taints []map[string]any `json:"taints"`
Ready string `json:"ready"`
} `json:"spec"`
Status struct {
Capacity map[string]string `json:"capacity"`
NodeInfo struct {
KubeletVersion string `json:"kubeletVersion"`
} `json:"nodeInfo"`
} `json:"status"`
}
func parseWayneNamespaceDetail(body []byte) (wayneNamespaceDetail, error) {
var wrapped struct {
Data wayneNamespaceDetail `json:"data"`
}
if err := json.Unmarshal(body, &wrapped); err != nil {
return wayneNamespaceDetail{}, err
}
return wrapped.Data, nil
}
func parseWayneApps(body []byte) ([]wayneApp, error) {
var wrapped struct {
Data struct {
List []wayneApp `json:"list"`
} `json:"data"`
}
if err := json.Unmarshal(body, &wrapped); err != nil {
return nil, err
}
return wrapped.Data.List, nil
}
func parseWayneDeployments(body []byte) ([]wayneDeployment, error) {
var wrapped struct {
Data struct {
List []wayneDeployment `json:"list"`
} `json:"data"`
}
if err := json.Unmarshal(body, &wrapped); err != nil {
return nil, err
}
return wrapped.Data.List, nil
}
func parseWaynePublishHistories(body []byte) ([]WayneDeploymentHistory, error) {
var wrapped struct {
Data struct {
List []WayneDeploymentHistory `json:"list"`
} `json:"data"`
}
if err := json.Unmarshal(body, &wrapped); err != nil {
return nil, err
}
for i := range wrapped.Data.List {
wrapped.Data.List[i].CreatedAt = parseWayneTime(wrapped.Data.List[i].CreateTime)
}
return wrapped.Data.List, nil
}
func parseWayneMapList(body []byte) ([]map[string]any, error) {
var wrapped struct {
Data struct {
List []map[string]any `json:"list"`
} `json:"data"`
}
if err := json.Unmarshal(body, &wrapped); err != nil {
return nil, err
}
return wrapped.Data.List, nil
}
func parseWayneTime(raw string) time.Time {
raw = strings.TrimSpace(raw)
if raw == "" {
return time.Time{}
}
layouts := []string{time.RFC3339, "2006-01-02T15:04:05Z07:00", "2006-01-02 15:04:05", "2006-01-02T15:04:05"}
for _, layout := range layouts {
if parsed, err := time.Parse(layout, raw); err == nil {
return parsed
}
}
return time.Time{}
}
func sortWayneDeploymentHistories(items []WayneDeploymentHistory) {
for i := 1; i < len(items); i++ {
item := items[i]
j := i - 1
for j >= 0 && items[j].CreatedAt.Before(item.CreatedAt) {
items[j+1] = items[j]
j--
}
items[j+1] = item
}
}
func parseWayneNodeSummary(body []byte) (wayneNodeData, error) {
var wrapped struct {
Data struct {
NodeSummary struct {
Total int `json:"total"`
Ready int `json:"ready"`
Schedulable int `json:"schedulable"`
} `json:"nodeSummary"`
CPUSummary struct {
Total float64 `json:"total"`
Used float64 `json:"used"`
} `json:"cpuSummary"`
MemorySummary struct {
Total float64 `json:"total"`
Used float64 `json:"used"`
} `json:"memorySummary"`
Nodes []wayneNode `json:"nodes"`
} `json:"data"`
}
if err := json.Unmarshal(body, &wrapped); err != nil {
return wayneNodeData{}, err
}
return wayneNodeData{
Total: wrapped.Data.NodeSummary.Total,
Ready: wrapped.Data.NodeSummary.Ready,
Schedulable: wrapped.Data.NodeSummary.Schedulable,
CPUUsed: wrapped.Data.CPUSummary.Used,
CPUTotal: wrapped.Data.CPUSummary.Total,
MemoryUsed: wrapped.Data.MemorySummary.Used,
MemoryTotal: wrapped.Data.MemorySummary.Total,
Nodes: wrapped.Data.Nodes,
}, nil
}
func containerClusterFromWayne(name string, data wayneNodeData) ContainerCluster {
status := "健康"
statusClass := "ok"
if data.Total == 0 {
status = "无节点"
statusClass = "idle"
} else if data.Ready < data.Total {
status = strconv.Itoa(data.Total-data.Ready) + " 节点告警"
statusClass = "warn"
}
cpu := ratioPercent(data.CPUUsed, data.CPUTotal)
version := "-"
for _, node := range data.Nodes {
if strings.TrimSpace(node.Status.NodeInfo.KubeletVersion) != "" {
version = strings.TrimSpace(node.Status.NodeInfo.KubeletVersion)
break
}
}
return ContainerCluster{
Name: name,
Zone: "-",
Status: status,
StatusClass: statusClass,
Nodes: data.Total,
ReadyNodes: data.Ready,
CPU: cpu,
CPUClass: warnClass(cpu),
Version: version,
Calico: "-",
}
}
func containerNodeFromWayne(cluster string, node wayneNode, businessLineName string) ContainerNode {
cpu := 0
if node.Status.Capacity != nil {
cpu, _ = strconv.Atoi(strings.TrimSpace(node.Status.Capacity["cpu"]))
}
memory := strings.TrimSpace(node.Status.Capacity["memory"])
spec := "-"
if cpu > 0 && memory != "" {
spec = fmt.Sprintf("%dC/%sG", cpu, memory)
} else if cpu > 0 {
spec = fmt.Sprintf("%dC", cpu)
} else if memory != "" {
spec = memory + "G"
}
ready := strings.EqualFold(strings.TrimSpace(node.Spec.Ready), "true")
status := "Ready"
statusClass := "ok"
if !ready {
status = "NotReady"
statusClass = "warn"
} else if node.Spec.Unschedulable {
status = "Ready · 不可调度"
statusClass = "warn"
}
label := businessLineLabel(node.Labels, businessLineName)
return ContainerNode{
Cluster: cluster,
Name: node.Name,
IP: firstNodeIP(node.Labels),
Label: label,
Taint: formatNodeTaints(node.Spec.Taints),
Spec: spec,
CPU: 0,
CPUClass: "",
Status: status,
StatusClass: statusClass,
}
}
func warnClass(value int) string {
if value >= 80 {
return "warn"
}
return ""
}
func businessLineLabel(labels map[string]string, businessLineName string) string {
for _, key := range []string{"business-line", "business_line", "xinfra/business-line"} {
if value := strings.TrimSpace(labels[key]); value != "" {
return key + "=" + value
}
}
if businessLineName != "" {
return "business-line=" + businessLineName
}
return "-"
}
func firstNodeIP(labels map[string]string) string {
for _, key := range []string{"kubernetes.io/hostname", "internal-ip", "node-ip"} {
if value := strings.TrimSpace(labels[key]); value != "" && strings.Contains(value, ".") {
return value
}
}
return "-"
}
func formatNodeTaints(taints []map[string]any) string {
if len(taints) == 0 {
return "-"
}
parts := make([]string, 0, len(taints))
for _, taint := range taints {
key := strings.TrimSpace(stringFromAny(taint["key"]))
value := strings.TrimSpace(stringFromAny(taint["value"]))
effect := strings.TrimSpace(stringFromAny(taint["effect"]))
text := key
if value != "" {
text += "=" + value
}
if effect != "" {
text += ":" + effect
}
if text != "" {
parts = append(parts, text)
}
}
if len(parts) == 0 {
return "-"
}
return strings.Join(parts, ", ")
}
func sortContainerClusters(items []ContainerCluster) {
for i := 1; i < len(items); i++ {
item := items[i]
j := i - 1
for j >= 0 && items[j].Name > item.Name {
items[j+1] = items[j]
j--
}
items[j+1] = item
}
}
func sortContainerNodes(items []ContainerNode) {
for i := 1; i < len(items); i++ {
item := items[i]
j := i - 1
for j >= 0 && (items[j].Cluster > item.Cluster || (items[j].Cluster == item.Cluster && items[j].Name > item.Name)) {
items[j+1] = items[j]
j--
}
items[j+1] = item
}
}
func containerWorkloadFromMap(item map[string]any, cluster string, namespace string, businessLineName string) ContainerWorkload {
name := firstString(item, "name", "objectMeta.name", "metadata.name")
ready := firstInt(item, "pods.current", "readyReplicas", "status.readyReplicas", "availableReplicas", "status.availableReplicas")
total := firstInt(item, "pods.desired", "replicas", "status.replicas", "desiredReplicas", "spec.replicas")
if total == 0 && ready > 0 {
total = ready
}
status := "健康"
statusClass := "ok"
if total == 0 {
status = "未就绪"
statusClass = "warn"
} else if ready < total {
status = "部分异常"
statusClass = "warn"
}
return ContainerWorkload{
Name: name,
Cluster: cluster,
Namespace: namespace,
Workload: "Deployment",
Ready: strconv.Itoa(ready) + " / " + strconv.Itoa(total),
ReadyPods: ready,
Pods: total,
Image: firstImage(item),
BusinessLine: businessLineName,
Status: status,
StatusClass: statusClass,
}
}
func firstImage(item map[string]any) string {
if value := firstString(item, "image", "containerImage"); value != "" {
return value
}
for _, path := range []string{"containers", "images", "containerImages"} {
value, ok := nestedValue(item, path)
if !ok {
continue
}
if images, ok := value.([]any); ok && len(images) > 0 {
if image, ok := images[0].(string); ok {
return image
}
}
}
if containers, ok := nestedValue(item, "spec.template.spec.containers"); ok {
if rows, ok := containers.([]any); ok && len(rows) > 0 {
if row, ok := rows[0].(map[string]any); ok {
return stringFromAny(row["image"])
}
}
}
return ""
}
func firstString(item map[string]any, paths ...string) string {
for _, path := range paths {
value, ok := nestedValue(item, path)
if !ok {
continue
}
if text := strings.TrimSpace(stringFromAny(value)); text != "" {
return text
}
}
return ""
}
func firstInt(item map[string]any, paths ...string) int {
for _, path := range paths {
value, ok := nestedValue(item, path)
if !ok {
continue
}
switch typed := value.(type) {
case float64:
return int(typed)
case int:
return typed
case json.Number:
intValue, _ := typed.Int64()
return int(intValue)
case string:
intValue, _ := strconv.Atoi(strings.TrimSpace(typed))
return intValue
}
}
return 0
}
func nestedValue(item map[string]any, path string) (any, bool) {
current := any(item)
for _, part := range strings.Split(path, ".") {
row, ok := current.(map[string]any)
if !ok {
return nil, false
}
current, ok = row[part]
if !ok {
return nil, false
}
}
return current, true
}
func stringFromAny(value any) string {
switch typed := value.(type) {
case string:
return typed
case fmt.Stringer:
return typed.String()
default:
if value == nil {
return ""
}
return fmt.Sprintf("%v", value)
}
}
func ratioPercent(used, total float64) int {
if total <= 0 {
return 0
}
return int(used * 100 / total)
}