opsx
5e59347c40
=== 0. 【严重】修复上一版引入的死锁（平台写路径全卡死）===
上一版把保留期读取写成 `GetRetentionSettings()`（内部 db.RLock()），
却在 `db.Lock()` 已持有的情况下调用（recordDelivery / RecordAudit 的裁剪路径）。
Go 的 sync.RWMutex **不可重入** → 第一次写投递历史/审计日志就永久死锁。
现象：登录 28 秒返回空（登录也要写审计）、所有写接口挂住。
go build / go vet / 之前的单测（不触发写路径）全都查不出来。
修法：
- 拆出 retentionFromDBLocked(db) / alertDeliveryRetentionLocked(db) /
auditLogRetentionLocked(db)，调用方已持锁时用无锁版本
- upsertSettingLocked 里**不能**用 db.NextID()（内部 d.mu.Lock()，同样死锁），
改为锁内自算 max(id)+1
- 新增 retention_test.go：持写锁调用裁剪函数，2 秒超时未返回即判死锁
实测修复后：登录 103ms（原 28000ms）、panic/deadlock 命中 0。
=== 1. 平台数据保留期可配（系统设置 → 通用配置 → 日志保留时间）===
新增 GET/PUT /api/v1/settings/retention：
alertDeliveryDays 告警发送历史保留天数，默认 7（原写死"最近 50 条"）
auditLogDays 审计日志保留天数，默认 90（原写死常量）
保存后立即按新策略裁剪一次，不必等下次触发。
校验：发送历史 1~365 天、审计 1~3650 天，超出返回明确中文原因（不是 500）。
注意与「Loki 日志保留期」区分：那个是部署级配置（组件部署 → 保留策略）。
=== 2. 发送历史：50 条硬上限 → 保留 7 天 ===
后端 Deliveries 接口原来写死 `len(out) < 50`，现在按保留期取（响应上限 500 可调），
标题改为「发送历史（保留 N 天）」，N 从设置里读。
=== 3. 菜单「日志服务」→「组件部署」===
=== 4. 分析页：去掉「全部命名空间」+ 改手动刷新 ===
- 移除「全部命名空间」选项：{namespace=~".+"} 要扫全集群（约 370GB/天），
一次分析几十秒，得不偿失；历史值 __all__ 也会被忽略并回退到策略/第一个 ns
- 筛选改为「选择值 / 已生效值」分离：切换命名空间或时间范围**不再自动查询**，
只把刷新按钮点亮并提示「筛选条件已改，点刷新生效」；
点一次才应用并取数。
原因：一次分析要跑多条 count_over_time，切一下下拉就重查一轮，
快速连点会把 Loki 打满（并发护栏 429），还拖慢别人。
=== 实测（9739c7c092e7）===
登录 103ms / token 273 字符
GET retention → {alertDeliveryDays:7, auditLogDays:90}
PUT 0 天 → 400「告警发送历史保留天数需在 1~365 之间」
平台日志 panic/deadlock 命中 0
go test ./... 全绿（含 2 个新死锁回归用例）
371 lines
12 KiB
Go
371 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"
|
|
"lokistack/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"
|
|
}
|