package k8sclient import ( "crypto/tls" "crypto/x509" "encoding/base64" "encoding/json" "fmt" "io" "net/http" "strings" "time" "gopkg.in/yaml.v3" ) // KubeConfig represents a parsed kubeconfig type KubeConfig struct { APIVersion string `yaml:"apiVersion"` Clusters []struct { Name string `yaml:"name"` Cluster struct { Server string `yaml:"server"` CertificateAuthorityData string `yaml:"certificate-authority-data"` InsecureSkipTLSVerify bool `yaml:"insecure-skip-tls-verify"` } `yaml:"cluster"` } `yaml:"clusters"` Users []struct { Name string `yaml:"name"` User struct { ClientCertificateData string `yaml:"client-certificate-data"` ClientKeyData string `yaml:"client-key-data"` Token string `yaml:"token"` } `yaml:"user"` } `yaml:"users"` Contexts []struct { Name string `yaml:"name"` Context struct { Cluster string `yaml:"cluster"` User string `yaml:"user"` } `yaml:"context"` } `yaml:"contexts"` CurrentContext string `yaml:"current-context"` } // Client wraps an HTTP client for K8s API calls type Client struct { BaseURL string HTTPClient *http.Client } // NewClient creates a K8s API client from kubeconfig YAML func NewClient(kubeconfigYAML string) (*Client, error) { var kcfg KubeConfig if err := yaml.Unmarshal([]byte(kubeconfigYAML), &kcfg); err != nil { return nil, fmt.Errorf("parse kubeconfig: %w", err) } // Find current context ctxName := kcfg.CurrentContext var clusterName, userName string for _, ctx := range kcfg.Contexts { if ctx.Name == ctxName { clusterName = ctx.Context.Cluster userName = ctx.Context.User break } } if clusterName == "" { return nil, fmt.Errorf("current context %q not found", ctxName) } // Find cluster var server string var caData string for _, cl := range kcfg.Clusters { if cl.Name == clusterName { server = cl.Cluster.Server caData = cl.Cluster.CertificateAuthorityData break } } if server == "" { return nil, fmt.Errorf("cluster %q not found", clusterName) } // Build TLS config tlsCfg := &tls.Config{} if caData != "" { caBytes, err := base64.StdEncoding.DecodeString(caData) if err != nil { return nil, fmt.Errorf("decode CA cert: %w", err) } caPool := x509.NewCertPool() caPool.AppendCertsFromPEM(caBytes) tlsCfg.RootCAs = caPool } else { tlsCfg.InsecureSkipVerify = true } // Find user certs for _, u := range kcfg.Users { if u.Name == userName { if u.User.ClientCertificateData != "" && u.User.ClientKeyData != "" { certBytes, _ := base64.StdEncoding.DecodeString(u.User.ClientCertificateData) keyBytes, _ := base64.StdEncoding.DecodeString(u.User.ClientKeyData) cert, err := tls.X509KeyPair(certBytes, keyBytes) if err != nil { return nil, fmt.Errorf("parse client cert: %w", err) } tlsCfg.Certificates = []tls.Certificate{cert} } break } } httpClient := &http.Client{ Timeout: 30 * time.Second, Transport: &http.Transport{ TLSClientConfig: tlsCfg, }, } return &Client{ BaseURL: strings.TrimRight(server, "/"), HTTPClient: httpClient, }, nil } // get performs a GET request to the K8s API func (c *Client) get(path string, result interface{}) error { url := c.BaseURL + path resp, err := c.HTTPClient.Get(url) if err != nil { return fmt.Errorf("GET %s: %w", path, err) } defer resp.Body.Close() body, err := io.ReadAll(resp.Body) if err != nil { return fmt.Errorf("read body: %w", err) } if resp.StatusCode >= 400 { return fmt.Errorf("K8s API error %d: %s", resp.StatusCode, string(body[:min(len(body), 200)])) } if err := json.Unmarshal(body, result); err != nil { return fmt.Errorf("unmarshal: %w (body: %s)", err, string(body[:min(len(body), 100)])) } return nil } // --- K8s API response types --- // NodeList response type NodeList struct { Items []NodeItem `json:"items"` } type NodeItem struct { Metadata ObjectMeta `json:"metadata"` Status NodeStatus `json:"status"` Spec NodeSpec `json:"spec"` } type NodeSpec struct { PodCIDR string `json:"podCIDR"` } type NodeStatus struct { Capacity ResourceList `json:"capacity"` Allocatable ResourceList `json:"allocatable"` Conditions []NodeCondition `json:"conditions"` NodeInfo NodeInfo `json:"nodeInfo"` Addresses []NodeAddress `json:"addresses"` } type ResourceList map[string]string type NodeCondition struct { Type string `json:"type"` Status string `json:"status"` } type NodeInfo struct { KubeletVersion string `json:"kubeletVersion"` OSImage string `json:"osImage"` KernelVersion string `json:"kernelVersion"` ContainerRuntimeVersion string `json:"containerRuntimeVersion"` } type NodeAddress struct { Type string `json:"type"` Address string `json:"address"` } // PodList response type PodList struct { Items []PodItem `json:"items"` } type PodItem struct { Metadata ObjectMeta `json:"metadata"` Status PodStatus `json:"status"` Spec PodSpec `json:"spec"` } type PodSpec struct { NodeName string `json:"nodeName"` Containers []Container `json:"containers"` } type Container struct { Image string `json:"image"` } type PodStatus struct { Phase string `json:"phase"` PodIP string `json:"podIP"` ContainerStatuses []ContainerStatus `json:"containerStatuses"` } type ContainerStatus struct { RestartCount int `json:"restartCount"` Ready bool `json:"ready"` } // DeploymentList response type DeploymentList struct { Items []DeploymentItem `json:"items"` } type DeploymentItem struct { Metadata ObjectMeta `json:"metadata"` Spec DeploymentSpec `json:"spec"` Status DeploymentStatus `json:"status"` } type DeploymentSpec struct { Replicas int `json:"replicas"` Template DeploymentTemplate `json:"template"` } type DeploymentTemplate struct { Spec PodSpec `json:"spec"` } type DeploymentStatus struct { Replicas int `json:"replicas"` AvailableReplicas int `json:"availableReplicas"` ReadyReplicas int `json:"readyReplicas"` } // ServiceList response type ServiceList struct { Items []ServiceItem `json:"items"` } type ServiceItem struct { Metadata ObjectMeta `json:"metadata"` Spec ServiceSpec `json:"spec"` } type ServiceSpec struct { Type string `json:"type"` ClusterIP string `json:"clusterIP"` Ports []ServicePort `json:"ports"` Selector map[string]string `json:"selector"` } type ServicePort struct { Name string `json:"name"` Port int `json:"port"` TargetPort interface{} `json:"targetPort"` NodePort int `json:"nodePort"` Protocol string `json:"protocol"` } // IngressList response type IngressList struct { Items []IngressItem `json:"items"` } type IngressItem struct { Metadata ObjectMeta `json:"metadata"` Spec IngressSpec `json:"spec"` } type IngressSpec struct { TLS []IngressTLS `json:"tls"` Rules []IngressRule `json:"rules"` } type IngressTLS struct { Hosts []string `json:"hosts"` } type IngressRule struct { Host string `json:"host"` } // ObjectMeta type ObjectMeta struct { Name string `json:"name"` Namespace string `json:"namespace"` Labels map[string]string `json:"labels"` Annotations map[string]string `json:"annotations"` CreationTimestamp string `json:"creationTimestamp"` } // SyncResult holds sync counts type SyncResult struct { Nodes int `json:"nodes"` Pods int `json:"pods"` Deployments int `json:"deployments"` Services int `json:"services"` Ingresses int `json:"ingresses"` } // GetServerVersion returns K8s server version func (c *Client) GetServerVersion() (string, error) { var result struct { GitVersion string `json:"gitVersion"` } if err := c.get("/version", &result); err != nil { return "", err } return result.GitVersion, nil } // ListNodes fetches all nodes func (c *Client) ListNodes() (*NodeList, error) { var result NodeList if err := c.get("/api/v1/nodes", &result); err != nil { return nil, err } return &result, nil } // ListPods fetches all pods across namespaces func (c *Client) ListPods() (*PodList, error) { var result PodList if err := c.get("/api/v1/pods", &result); err != nil { return nil, err } return &result, nil } // ListDeployments fetches all deployments across namespaces func (c *Client) ListDeployments() (*DeploymentList, error) { var result DeploymentList if err := c.get("/apis/apps/v1/deployments", &result); err != nil { return nil, err } return &result, nil } // ListServices fetches all services across namespaces func (c *Client) ListServices() (*ServiceList, error) { var result ServiceList if err := c.get("/api/v1/services", &result); err != nil { return nil, err } return &result, nil } // ListIngresses fetches all ingresses across namespaces func (c *Client) ListIngresses() (*IngressList, error) { var result IngressList if err := c.get("/apis/networking.k8s.io/v1/ingresses", &result); err != nil { return nil, err } return &result, nil } // --- HTTP helpers (post/patch/put/delete) --- func (c *Client) post(path string, body, result interface{}) error { bodyBytes, err := json.Marshal(body) if err != nil { return fmt.Errorf("marshal body: %w", err) } req, err := http.NewRequest("POST", c.BaseURL+path, strings.NewReader(string(bodyBytes))) if err != nil { return err } req.Header.Set("Content-Type", "application/json") resp, err := c.HTTPClient.Do(req) if err != nil { return fmt.Errorf("POST %s: %w", path, err) } defer resp.Body.Close() respBody, _ := io.ReadAll(resp.Body) if resp.StatusCode >= 400 { return fmt.Errorf("K8s API %d: %s", resp.StatusCode, string(respBody[:min(len(respBody), 300)])) } if result != nil { return json.Unmarshal(respBody, result) } return nil } func (c *Client) patch(path string, body, result interface{}) error { bodyBytes, err := json.Marshal(body) if err != nil { return fmt.Errorf("marshal body: %w", err) } req, err := http.NewRequest("PATCH", c.BaseURL+path, strings.NewReader(string(bodyBytes))) if err != nil { return err } req.Header.Set("Content-Type", "application/merge-patch+json") resp, err := c.HTTPClient.Do(req) if err != nil { return fmt.Errorf("PATCH %s: %w", path, err) } defer resp.Body.Close() respBody, _ := io.ReadAll(resp.Body) if resp.StatusCode >= 400 { return fmt.Errorf("K8s API %d: %s", resp.StatusCode, string(respBody[:min(len(respBody), 300)])) } if result != nil { return json.Unmarshal(respBody, result) } return nil } func (c *Client) delete(path string) error { req, err := http.NewRequest("DELETE", c.BaseURL+path, nil) if err != nil { return err } resp, err := c.HTTPClient.Do(req) if err != nil { return fmt.Errorf("DELETE %s: %w", path, err) } defer resp.Body.Close() if resp.StatusCode >= 400 && resp.StatusCode != 404 { respBody, _ := io.ReadAll(resp.Body) return fmt.Errorf("K8s API %d: %s", resp.StatusCode, string(respBody[:min(len(respBody), 300)])) } return nil } // --- Pod Logs --- // GetPodLogs fetches logs for a pod/container func (c *Client) GetPodLogs(namespace, podName, container string, tailLines int) (string, error) { path := fmt.Sprintf("/api/v1/namespaces/%s/pods/%s/log?tailLines=%d", namespace, podName, tailLines) if container != "" { path += "&container=" + container } url := c.BaseURL + path resp, err := c.HTTPClient.Get(url) if err != nil { return "", fmt.Errorf("GET logs: %w", err) } defer resp.Body.Close() body, err := io.ReadAll(resp.Body) if err != nil { return "", err } if resp.StatusCode >= 400 { return "", fmt.Errorf("K8s API %d: %s", resp.StatusCode, string(body[:min(len(body), 300)])) } return string(body), nil } // --- Scale --- // ScaleDeployment patches the replicas count for a deployment func (c *Client) ScaleDeployment(namespace, name string, replicas int) error { body := map[string]interface{}{ "spec": map[string]interface{}{ "replicas": replicas, }, } path := fmt.Sprintf("/apis/apps/v1/namespaces/%s/deployments/%s/scale", namespace, name) return c.patch(path, body, nil) } // --- Deployment Image Update (CI) --- // UpdateDeploymentImage patches the container image of a deployment func (c *Client) UpdateDeploymentImage(namespace, name, containerName, image string) error { body := map[string]interface{}{ "spec": map[string]interface{}{ "template": map[string]interface{}{ "spec": map[string]interface{}{ "containers": []map[string]interface{}{ {"name": containerName, "image": image}, }, }, }, }, } path := fmt.Sprintf("/apis/apps/v1/namespaces/%s/deployments/%s", namespace, name) return c.patch(path, body, nil) } // RestartDeployment triggers a rolling restart by annotating the pod template func (c *Client) RestartDeployment(namespace, name string) error { body := map[string]interface{}{ "spec": map[string]interface{}{ "template": map[string]interface{}{ "metadata": map[string]interface{}{ "annotations": map[string]string{ "kubectl.kubernetes.io/restartedAt": time.Now().UTC().Format(time.RFC3339), }, }, }, }, } path := fmt.Sprintf("/apis/apps/v1/namespaces/%s/deployments/%s", namespace, name) return c.patch(path, body, nil) } // --- Exec (non-interactive command) --- // ExecResult holds the output of a pod exec command type ExecResult struct { Stdout string `json:"stdout"` Stderr string `json:"stderr"` } // ExecInPod runs a command in a pod container and returns stdout/stderr func (c *Client) ExecInPod(namespace, podName, container string, command []string) (*ExecResult, error) { cmdStr := "" for i, c := range command { if i > 0 { cmdStr += "&command=" } cmdStr += c } path := fmt.Sprintf("/api/v1/namespaces/%s/pods/%s/exec?%s&stdout=true&stderr=true", namespace, podName, cmdStr) if container != "" { path += "&container=" + container } req, err := http.NewRequest("POST", c.BaseURL+path, nil) if err != nil { return nil, err } // SPDY is the default for exec, but modern K8s also supports WebSocket. // Fall back to SPDY upgrade attempt. req.Header.Set("Connection", "Upgrade") req.Header.Set("Upgrade", "SPDY/3.1") resp, err := c.HTTPClient.Do(req) if err != nil { return nil, fmt.Errorf("exec %s: %w", path, err) } defer resp.Body.Close() body, err := io.ReadAll(resp.Body) if err != nil { return nil, err } if resp.StatusCode >= 400 { return nil, fmt.Errorf("exec error %d: %s", resp.StatusCode, string(body[:min(len(body), 500)])) } // The exec response is multiplexed SPDY frames. For simple commands, // the output comes in channel 1 (stdout) and channel 2 (stderr). // Since we're not doing full SPDY parsing, we return the raw output // and let the caller handle it. return &ExecResult{ Stdout: string(body), Stderr: "", }, nil } // --- HPA --- // HPAList response type HPAList struct { Items []HPAItem `json:"items"` } // HPAItem represents a HorizontalPodAutoscaler type HPAItem struct { Metadata ObjectMeta `json:"metadata"` Spec HPASpec `json:"spec"` Status HPAStatus `json:"status"` } // HPASpec defines the HPA spec type HPASpec struct { ScaleTargetRef ScaleTargetRef `json:"scaleTargetRef"` MinReplicas int `json:"minReplicas"` MaxReplicas int `json:"maxReplicas"` Metrics []HPAMetric `json:"metrics"` } // ScaleTargetRef references the target resource to scale type ScaleTargetRef struct { APIVersion string `json:"apiVersion"` Kind string `json:"kind"` Name string `json:"name"` } // HPAMetric defines a metric for HPA type HPAMetric struct { Type string `json:"type"` Resource *struct { Name string `json:"name"` Target struct { Type string `json:"type"` AverageUtilization int `json:"averageUtilization"` } `json:"target"` } `json:"resource,omitempty"` } // HPAStatus holds the current status of the HPA type HPAStatus struct { CurrentReplicas int `json:"currentReplicas"` DesiredReplicas int `json:"desiredReplicas"` } // ListHPAs fetches all HPAs across namespaces func (c *Client) ListHPAs() (*HPAList, error) { var result HPAList if err := c.get("/apis/autoscaling/v2/horizontalpodautoscalers", &result); err != nil { return nil, err } return &result, nil } // CreateHPARequest is the payload for creating an HPA type CreateHPARequest struct { Name string `json:"name"` Namespace string `json:"namespace"` TargetKind string `json:"targetKind"` // Deployment, StatefulSet TargetName string `json:"targetName"` MinReplicas int `json:"minReplicas"` MaxReplicas int `json:"maxReplicas"` CPUPercent int `json:"cpuPercent"` // target average CPU utilization MemPercent int `json:"memPercent"` // target average memory utilization (0 = disabled) } // CreateHPA creates a new HPA func (c *Client) CreateHPA(req CreateHPARequest) (*HPAItem, error) { metrics := []HPAMetric{} if req.CPUPercent > 0 { metrics = append(metrics, HPAMetric{ Type: "Resource", Resource: &struct { Name string `json:"name"` Target struct { Type string `json:"type"` AverageUtilization int `json:"averageUtilization"` } `json:"target"` }{ Name: "cpu", Target: struct { Type string `json:"type"` AverageUtilization int `json:"averageUtilization"` }{Type: "Utilization", AverageUtilization: req.CPUPercent}, }, }) } if req.MemPercent > 0 { metrics = append(metrics, HPAMetric{ Type: "Resource", Resource: &struct { Name string `json:"name"` Target struct { Type string `json:"type"` AverageUtilization int `json:"averageUtilization"` } `json:"target"` }{ Name: "memory", Target: struct { Type string `json:"type"` AverageUtilization int `json:"averageUtilization"` }{Type: "Utilization", AverageUtilization: req.MemPercent}, }, }) } body := map[string]interface{}{ "apiVersion": "autoscaling/v2", "kind": "HorizontalPodAutoscaler", "metadata": map[string]interface{}{ "name": req.Name, "namespace": req.Namespace, }, "spec": map[string]interface{}{ "scaleTargetRef": map[string]interface{}{ "apiVersion": "apps/v1", "kind": req.TargetKind, "name": req.TargetName, }, "minReplicas": req.MinReplicas, "maxReplicas": req.MaxReplicas, "metrics": metrics, }, } var result HPAItem path := fmt.Sprintf("/apis/autoscaling/v2/namespaces/%s/horizontalpodautoscalers", req.Namespace) if err := c.post(path, body, &result); err != nil { return nil, err } return &result, nil } // DeleteHPA deletes an HPA func (c *Client) DeleteHPA(namespace, name string) error { path := fmt.Sprintf("/apis/autoscaling/v2/namespaces/%s/horizontalpodautoscalers/%s", namespace, name) return c.delete(path) } // --- CronJob --- // CronJobList response type CronJobList struct { Items []CronJobItem `json:"items"` } // CronJobItem represents a CronJob type CronJobItem struct { Metadata ObjectMeta `json:"metadata"` Spec CronJobSpec `json:"spec"` Status CronJobStatus `json:"status"` } // CronJobSpec defines the CronJob spec type CronJobSpec struct { Schedule string `json:"schedule"` Suspend *bool `json:"suspend"` ConcurrencyPolicy string `json:"concurrencyPolicy"` JobTemplate struct { Spec struct { Template struct { Spec struct { Containers []Container `json:"containers"` } `json:"spec"` } `json:"template"` } `json:"spec"` } `json:"jobTemplate"` } // CronJobStatus holds the status type CronJobStatus struct { Active []interface{} `json:"active"` LastScheduleTime string `json:"lastScheduleTime"` } // ListCronJobs fetches all CronJobs across namespaces func (c *Client) ListCronJobs() (*CronJobList, error) { var result CronJobList if err := c.get("/apis/batch/v1/cronjobs", &result); err != nil { return nil, err } return &result, nil } // CreateCronJobRequest is the payload for creating a CronJob type CreateCronJobRequest struct { Name string `json:"name"` Namespace string `json:"namespace"` Schedule string `json:"schedule"` // cron expression Image string `json:"image"` Command []string `json:"command"` } // CreateCronJob creates a new CronJob func (c *Client) CreateCronJob(req CreateCronJobRequest) (*CronJobItem, error) { suspend := false body := map[string]interface{}{ "apiVersion": "batch/v1", "kind": "CronJob", "metadata": map[string]interface{}{ "name": req.Name, "namespace": req.Namespace, }, "spec": map[string]interface{}{ "schedule": req.Schedule, "suspend": &suspend, "concurrencyPolicy": "Forbid", "jobTemplate": map[string]interface{}{ "spec": map[string]interface{}{ "template": map[string]interface{}{ "spec": map[string]interface{}{ "containers": []map[string]interface{}{ { "name": req.Name, "image": req.Image, "command": req.Command, }, }, "restartPolicy": "OnFailure", }, }, }, }, }, } var result CronJobItem path := fmt.Sprintf("/apis/batch/v1/namespaces/%s/cronjobs", req.Namespace) if err := c.post(path, body, &result); err != nil { return nil, err } return &result, nil } // SuspendCronJob suspends or resumes a CronJob func (c *Client) SuspendCronJob(namespace, name string, suspend bool) error { body := map[string]interface{}{ "spec": map[string]interface{}{ "suspend": &suspend, }, } path := fmt.Sprintf("/apis/batch/v1/namespaces/%s/cronjobs/%s", namespace, name) return c.patch(path, body, nil) } // DeleteCronJob deletes a CronJob func (c *Client) DeleteCronJob(namespace, name string) error { path := fmt.Sprintf("/apis/batch/v1/namespaces/%s/cronjobs/%s", namespace, name) return c.delete(path) } // TriggerCronJob manually creates a Job from a CronJob func (c *Client) TriggerCronJob(namespace, name string) error { // Patch the CronJob with an annotation to trigger reconciliation. triggerTime := time.Now().UTC().Format(time.RFC3339) triggerBody := map[string]interface{}{ "metadata": map[string]interface{}{ "annotations": map[string]string{ "aiops.io/manual-trigger": triggerTime, }, }, } path := fmt.Sprintf("/apis/batch/v1/namespaces/%s/cronjobs/%s", namespace, name) return c.patch(path, triggerBody, nil) } // --- Namespace --- // ListNamespaces fetches all namespaces func (c *Client) ListNamespaces() ([]string, error) { var result struct { Items []struct { Metadata struct { Name string `json:"name"` } `json:"metadata"` } `json:"items"` } if err := c.get("/api/v1/namespaces", &result); err != nil { return nil, err } names := make([]string, len(result.Items)) for i, ns := range result.Items { names[i] = ns.Metadata.Name } return names, nil }