opsx
02d1bd87a7
=== 一、致命 bug：Promtail 清单第 10 个 %s 缺参数（导致日志采集中断）=== go vet 报："format %s reads arg #10, but call has 9 args"。 模板里 DaemonSet 的 `namespace: %s`（第 9 个）和 `image: %s`（第 10 个） 是两个独立占位符，但 Sprintf 只传了 9 个参数，第 10 个变成 %!s(MISSING)， 整个多文档 YAML 解析失败： "解析 Kubernetes YAML: yaml: line 23: found character that cannot start any token" 而 installDaemonset 是「先删 daemonset/configmap/serviceaccount 再 apply」， 所以每次失败都会把上一份 Promtail 删掉却不重建 → 日志采集中断。 修复： 1) Sprintf 补上 DaemonSet 命名空间参数（第 9 位） 2) 新增 validateManifest()：在删旧资源之前先本地解析校验清单， 校验不过直接中止、不碰任何现有资源 —— 让模板错误永远不会再打断采集 3) 新增回归测试 promtail_manifest_test.go：用平台同一套 YAMLOrJSONDecoder 解析 3 种参数组合，断言各解析出 5 个对象且无 TAB === 二、快捷 LogQL：Ingress 访问日志不再跟随所选命名空间 === 原实现：next = applyLabelFilter(next,'namespace', namespace.trim() || 'ingress-nginx') → 选了 monitor 就生成 {namespace="monitor"} |= "HTTP/"，查不到任何日志 （ingress controller 的 Pod 只在 ingress-nginx 命名空间里） 修复：ingress 模板独立处理 - 固定用 ingress controller 命名空间（ingress-nginx，缺失则回退 ingress） - 不叠加当前选中的 app / pod 过滤（那些属于业务服务） - 结果是 {namespace="ingress-nginx"} |= "HTTP/" === 三、K8s 资源名去掉 opsxloki === ClusterRole / ClusterRoleBinding：opsxloki-promtail → loki-promtail - PromtailDaemonsetYAML 里 3 处 name 全部改名 - uninstallDaemonset 同步改名，并保留对旧名 opsxloki-promtail 的兼容清理， 避免升级后留下孤儿 ClusterRole - 集群侧已实测：新建 loki-promtail，旧的 opsxloki-promtail 已删除 保留的 opsxloki（有兼容性约束，改动会丢数据/掉登录）： - Go module 导入路径 opsxloki/xxx（149 处，纯内部包名） - 数据文件 opsxloki.json（改了读不到历史数据） - 前端 localStorage key opsxloki.auth.* / opsxloki.theme.* 等（改了在线用户掉登录） - Dockerfile 里的 opsxloki 系统用户/二进制名 镜像：lghost/lokistack:latest + v0.1 = sha256:4a5590580cbb
373 lines
12 KiB
Go
373 lines
12 KiB
Go
package services
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"strings"
|
|
"time"
|
|
|
|
corev1 "k8s.io/api/core/v1"
|
|
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
|
"k8s.io/apimachinery/pkg/types"
|
|
"k8s.io/apimachinery/pkg/util/wait"
|
|
yamlutil "k8s.io/apimachinery/pkg/util/yaml"
|
|
"k8s.io/client-go/discovery"
|
|
"k8s.io/client-go/dynamic"
|
|
kcScheme "k8s.io/client-go/kubernetes/scheme"
|
|
"k8s.io/client-go/restmapper"
|
|
"k8s.io/client-go/tools/remotecommand"
|
|
"opsxloki/models"
|
|
)
|
|
|
|
func ensureNamespace(ctx context.Context, cluster *models.Cluster, namespace string) error {
|
|
client, err := clusterClient(cluster, 30*time.Second)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = client.CoreV1().Namespaces().Get(ctx, namespace, metav1.GetOptions{})
|
|
if apierrors.IsNotFound(err) {
|
|
_, err = client.CoreV1().Namespaces().Create(ctx, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}, metav1.CreateOptions{})
|
|
}
|
|
if apierrors.IsForbidden(err) {
|
|
// 命名空间级 Token 无权读取 Namespace 对象：命名空间由管理员预建，直接继续
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
|
|
// validateManifest 只做本地 YAML 解析校验，不访问集群。
|
|
//
|
|
// 用途：采集器 install 流程是「先删旧资源再 apply」，一旦模板生成有误
|
|
// （例如 fmt.Sprintf 占位符错位产生 %!s(MISSING)），就会出现
|
|
// 「旧的已删、新的没建成」→ 日志采集中断。所以在删之前先跑一遍解析。
|
|
func validateManifest(manifest string) error {
|
|
decoder := yamlutil.NewYAMLOrJSONDecoder(strings.NewReader(manifest), 4096)
|
|
count := 0
|
|
for {
|
|
var raw map[string]interface{}
|
|
if err := decoder.Decode(&raw); err == io.EOF {
|
|
break
|
|
} else if err != nil {
|
|
return fmt.Errorf("解析 Kubernetes YAML: %w", err)
|
|
}
|
|
if len(raw) == 0 {
|
|
continue
|
|
}
|
|
obj := &unstructured.Unstructured{Object: raw}
|
|
if obj.GetKind() == "" || obj.GetAPIVersion() == "" {
|
|
return fmt.Errorf("清单缺少 apiVersion/kind")
|
|
}
|
|
if obj.GetName() == "" {
|
|
return fmt.Errorf("Kubernetes %s 缺少 metadata.name", obj.GetKind())
|
|
}
|
|
count++
|
|
}
|
|
if count == 0 {
|
|
return fmt.Errorf("清单为空")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// applyManifest 使用 API Server 的 server-side apply，保留已有 YAML 模板与字段所有权。
|
|
func applyManifest(ctx context.Context, cluster *models.Cluster, manifest string) error {
|
|
cfg, err := clusterRESTConfig(cluster, 30*time.Second)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
dc, err := discovery.NewDiscoveryClientForConfig(cfg)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
groups, err := restmapper.GetAPIGroupResources(dc)
|
|
if err != nil {
|
|
return fmt.Errorf("读取 Kubernetes API 资源列表: %w", err)
|
|
}
|
|
mapper := restmapper.NewDiscoveryRESTMapper(groups)
|
|
client, err := dynamic.NewForConfig(cfg)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
decoder := yamlutil.NewYAMLOrJSONDecoder(strings.NewReader(manifest), 4096)
|
|
for {
|
|
var raw map[string]interface{}
|
|
if err := decoder.Decode(&raw); err == io.EOF {
|
|
break
|
|
} else if err != nil {
|
|
return fmt.Errorf("解析 Kubernetes YAML: %w", err)
|
|
}
|
|
if len(raw) == 0 {
|
|
continue
|
|
}
|
|
obj := &unstructured.Unstructured{Object: raw}
|
|
mapping, err := mapper.RESTMapping(obj.GroupVersionKind().GroupKind(), obj.GroupVersionKind().Version)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
name := obj.GetName()
|
|
if name == "" {
|
|
return fmt.Errorf("Kubernetes %s 缺少 metadata.name", obj.GetKind())
|
|
}
|
|
data, err := obj.MarshalJSON()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var resource dynamic.ResourceInterface = client.Resource(mapping.Resource)
|
|
if mapping.Scope.Name() == "namespace" {
|
|
resource = client.Resource(mapping.Resource).Namespace(obj.GetNamespace())
|
|
}
|
|
force := true
|
|
if _, err := resource.Patch(ctx, name, types.ApplyPatchType, data, metav1.PatchOptions{FieldManager: "lokideploy", Force: &force}); err != nil {
|
|
// 命名空间级 Token 无权 patch Namespace：视为已由管理员预建，跳过
|
|
if obj.GetKind() == "Namespace" && apierrors.IsForbidden(err) {
|
|
continue
|
|
}
|
|
return fmt.Errorf("应用 %s %s/%s: %w", obj.GetKind(), obj.GetNamespace(), name, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func waitStatefulSet(ctx context.Context, cluster *models.Cluster, namespace, name string) error {
|
|
client, err := clusterClient(cluster, 30*time.Second)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
limit, cancel := context.WithTimeout(ctx, 5*time.Minute)
|
|
defer cancel()
|
|
return wait.PollUntilContextCancel(limit, 3*time.Second, true, func(ctx context.Context) (bool, error) {
|
|
item, err := client.AppsV1().StatefulSets(namespace).Get(ctx, name, metav1.GetOptions{})
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
desired := int32(1)
|
|
if item.Spec.Replicas != nil {
|
|
desired = *item.Spec.Replicas
|
|
}
|
|
return item.Status.ObservedGeneration >= item.Generation && item.Status.ReadyReplicas >= desired && item.Status.UpdatedReplicas >= desired, nil
|
|
})
|
|
}
|
|
|
|
func waitDeployment(ctx context.Context, cluster *models.Cluster, namespace, name string) error {
|
|
client, err := clusterClient(cluster, 30*time.Second)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
limit, cancel := context.WithTimeout(ctx, 5*time.Minute)
|
|
defer cancel()
|
|
return wait.PollUntilContextCancel(limit, 3*time.Second, true, func(ctx context.Context) (bool, error) {
|
|
item, err := client.AppsV1().Deployments(namespace).Get(ctx, name, metav1.GetOptions{})
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
desired := int32(1)
|
|
if item.Spec.Replicas != nil {
|
|
desired = *item.Spec.Replicas
|
|
}
|
|
return item.Status.ObservedGeneration >= item.Generation && item.Status.ReadyReplicas >= desired && item.Status.UpdatedReplicas >= desired, nil
|
|
})
|
|
}
|
|
|
|
func waitDaemonSet(ctx context.Context, cluster *models.Cluster, namespace, name string) error {
|
|
client, err := clusterClient(cluster, 30*time.Second)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
limit, cancel := context.WithTimeout(ctx, 5*time.Minute)
|
|
defer cancel()
|
|
return wait.PollUntilContextCancel(limit, 3*time.Second, true, func(ctx context.Context) (bool, error) {
|
|
item, err := client.AppsV1().DaemonSets(namespace).Get(ctx, name, metav1.GetOptions{})
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
return item.Status.ObservedGeneration >= item.Generation && item.Status.NumberReady >= item.Status.DesiredNumberScheduled && item.Status.UpdatedNumberScheduled >= item.Status.DesiredNumberScheduled, nil
|
|
})
|
|
}
|
|
|
|
func restartDeployment(ctx context.Context, cluster *models.Cluster, namespace, name string) error {
|
|
client, err := clusterClient(cluster, 30*time.Second)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
item, err := client.AppsV1().Deployments(namespace).Get(ctx, name, metav1.GetOptions{})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
item.Spec.Template.Annotations = copyAnnotations(item.Spec.Template.Annotations)
|
|
item.Spec.Template.Annotations["kubectl.kubernetes.io/restartedAt"] = time.Now().UTC().Format(time.RFC3339)
|
|
_, err = client.AppsV1().Deployments(namespace).Update(ctx, item, metav1.UpdateOptions{})
|
|
return err
|
|
}
|
|
func copyAnnotations(source map[string]string) map[string]string {
|
|
result := map[string]string{}
|
|
for k, v := range source {
|
|
result[k] = v
|
|
}
|
|
return result
|
|
}
|
|
|
|
func deleteK8sResource(ctx context.Context, cluster *models.Cluster, kind, namespace, name string) error {
|
|
client, err := clusterClient(cluster, 30*time.Second)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var result error
|
|
switch kind {
|
|
case "statefulset":
|
|
result = client.AppsV1().StatefulSets(namespace).Delete(ctx, name, metav1.DeleteOptions{})
|
|
case "deployment":
|
|
result = client.AppsV1().Deployments(namespace).Delete(ctx, name, metav1.DeleteOptions{})
|
|
case "daemonset":
|
|
result = client.AppsV1().DaemonSets(namespace).Delete(ctx, name, metav1.DeleteOptions{})
|
|
case "service":
|
|
result = client.CoreV1().Services(namespace).Delete(ctx, name, metav1.DeleteOptions{})
|
|
case "configmap":
|
|
result = client.CoreV1().ConfigMaps(namespace).Delete(ctx, name, metav1.DeleteOptions{})
|
|
case "serviceaccount":
|
|
result = client.CoreV1().ServiceAccounts(namespace).Delete(ctx, name, metav1.DeleteOptions{})
|
|
case "persistentvolumeclaim":
|
|
result = client.CoreV1().PersistentVolumeClaims(namespace).Delete(ctx, name, metav1.DeleteOptions{})
|
|
case "clusterrole":
|
|
result = client.RbacV1().ClusterRoles().Delete(ctx, name, metav1.DeleteOptions{})
|
|
case "clusterrolebinding":
|
|
result = client.RbacV1().ClusterRoleBindings().Delete(ctx, name, metav1.DeleteOptions{})
|
|
default:
|
|
return fmt.Errorf("未知 Kubernetes 资源: %s", kind)
|
|
}
|
|
if apierrors.IsNotFound(result) {
|
|
return nil
|
|
}
|
|
return result
|
|
}
|
|
|
|
func serviceClusterIP(ctx context.Context, cluster *models.Cluster, namespace, name string) (string, error) {
|
|
client, err := clusterClient(cluster, 30*time.Second)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
svc, err := client.CoreV1().Services(namespace).Get(ctx, name, metav1.GetOptions{})
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return svc.Spec.ClusterIP, nil
|
|
}
|
|
func serviceNodePort(ctx context.Context, cluster *models.Cluster, namespace, name string) (int32, error) {
|
|
client, err := clusterClient(cluster, 30*time.Second)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
svc, err := client.CoreV1().Services(namespace).Get(ctx, name, metav1.GetOptions{})
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if len(svc.Spec.Ports) == 0 {
|
|
return 0, fmt.Errorf("Service %s/%s 没有端口", namespace, name)
|
|
}
|
|
return svc.Spec.Ports[0].NodePort, nil
|
|
}
|
|
|
|
func firstNodeIP(ctx context.Context, cluster *models.Cluster) (string, error) {
|
|
client, err := clusterClient(cluster, 30*time.Second)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
nodes, err := client.CoreV1().Nodes().List(ctx, metav1.ListOptions{Limit: 1})
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
for _, node := range nodes.Items {
|
|
for _, address := range node.Status.Addresses {
|
|
if address.Type == corev1.NodeInternalIP {
|
|
return address.Address, nil
|
|
}
|
|
}
|
|
}
|
|
return "", fmt.Errorf("集群无可用节点内网地址")
|
|
}
|
|
|
|
func patchDeployment(ctx context.Context, cluster *models.Cluster, namespace, name, yamlPatch string) error {
|
|
client, err := clusterClient(cluster, 30*time.Second)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
patch, err := yamlutil.ToJSON([]byte(yamlPatch))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = client.AppsV1().Deployments(namespace).Patch(ctx, name, types.StrategicMergePatchType, patch, metav1.PatchOptions{})
|
|
return err
|
|
}
|
|
|
|
|
|
// podDNSBroken 在指定 deployment 的容器里探测 hostname 是否可解析(wget /ready 连通性)。
|
|
func podDNSBroken(ctx context.Context, cluster *models.Cluster, namespace, deployment, container, host string) bool {
|
|
pods, err := listDeploymentPods(ctx, cluster, namespace, deployment)
|
|
if err != nil || len(pods) == 0 {
|
|
return false // 拿不到 pod 时不做自愈,交给后续日志排查
|
|
}
|
|
client, err := clusterClient(cluster, 15*time.Second)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
restCfg, err := clusterRESTConfig(cluster, 15*time.Second)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
execReq := client.CoreV1().RESTClient().Post().
|
|
Namespace(namespace).Resource("pods").Name(pods[0]).
|
|
SubResource("exec").
|
|
VersionedParams(&corev1.PodExecOptions{
|
|
Command: []string{"sh", "-c", "wget -q -O /dev/null -T 3 http://" + host + ":3100/ready"},
|
|
Container: container,
|
|
Stdout: true, Stderr: true,
|
|
}, kcScheme.ParameterCodec)
|
|
exec, err := remotecommand.NewSPDYExecutor(restCfg, "POST", execReq.URL())
|
|
if err != nil {
|
|
return false
|
|
}
|
|
var out bytes.Buffer
|
|
err = exec.StreamWithContext(ctx, remotecommand.StreamOptions{Stdout: &out, Stderr: &out})
|
|
return err != nil // 执行失败(连接/解析不通)视为 DNS 故障
|
|
}
|
|
|
|
// listDeploymentPods 按 deployment selector 列出 pod 名。
|
|
func listDeploymentPods(ctx context.Context, cluster *models.Cluster, namespace, deployment string) ([]string, error) {
|
|
client, err := clusterClient(cluster, 15*time.Second)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
dep, err := client.AppsV1().Deployments(namespace).Get(ctx, deployment, metav1.GetOptions{})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
selector := metav1.FormatLabelSelector(dep.Spec.Selector)
|
|
podList, err := client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: selector})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
names := make([]string, 0, len(podList.Items))
|
|
for i := range podList.Items {
|
|
names = append(names, podList.Items[i].Name)
|
|
}
|
|
return names, nil
|
|
}
|
|
|
|
|
|
// lokiNamespaceOf 从 endpoint 的 svc FQDN 提取 Loki 命名空间(log.ns.svc → ns)。
|
|
func lokiNamespaceOf(loki *models.LokiDeployment, endpoint string) string {
|
|
if loki != nil && loki.Namespace != "" {
|
|
return loki.Namespace
|
|
}
|
|
parts := strings.SplitN(endpoint, "//", 2)
|
|
if len(parts) == 2 {
|
|
host := strings.SplitN(parts[1], ".", 3)
|
|
if len(host) >= 2 {
|
|
return host[1]
|
|
}
|
|
}
|
|
return "log"
|
|
}
|