mirror of
https://github.com/labring/sealos.git
synced 2026-09-01 04:58:47 +08:00
feat(service): add iamge monitor to watch pull image error and slow pull (#6100)
* feat(service): add iamge monitor to watch pull image error and slow pull * fix: replace request * fix: ci lint * fix: comments convert to en
This commit is contained in:
zijiren
+10
-7
@@ -4,21 +4,24 @@ use (
|
||||
.
|
||||
./account
|
||||
./database
|
||||
./devbox
|
||||
./exceptionmonitor
|
||||
./hubble
|
||||
./imagemonitor
|
||||
./launchpad
|
||||
./pay
|
||||
./devbox
|
||||
./vlogs
|
||||
./hubble
|
||||
)
|
||||
|
||||
replace (
|
||||
github.com/google/gnostic-models => github.com/google/gnostic-models v0.6.8
|
||||
github.com/labring/sealos/controllers/account => ../controllers/account
|
||||
github.com/labring/sealos/controllers/user => ../controllers/user
|
||||
|
||||
k8s.io/api => k8s.io/api v0.32.3
|
||||
k8s.io/apiextensions-apiserver => k8s.io/apiextensions-apiserver v0.32.1
|
||||
k8s.io/apimachinery => k8s.io/apimachinery v0.32.1
|
||||
k8s.io/client-go => k8s.io/client-go v0.32.1
|
||||
sigs.k8s.io/controller-runtime => sigs.k8s.io/controller-runtime v0.20.4
|
||||
k8s.io/api => k8s.io/api v0.32.1
|
||||
k8s.io/apiextensions-apiserver => k8s.io/apiextensions-apiserver v0.32.1
|
||||
k8s.io/apimachinery => k8s.io/apimachinery v0.32.1
|
||||
k8s.io/client-go => k8s.io/client-go v0.32.1
|
||||
k8s.io/kube-openapi => k8s.io/kube-openapi v0.0.0-20241105132330-32ad38e42d3f
|
||||
sigs.k8s.io/controller-runtime => sigs.k8s.io/controller-runtime v0.20.4
|
||||
)
|
||||
|
||||
+328
-15
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,3 @@
|
||||
/.idea
|
||||
/imagemonitor
|
||||
/imagemonitor.exe
|
||||
@@ -0,0 +1,126 @@
|
||||
# Image Monitor Service
|
||||
|
||||
A real-time Kubernetes image pull monitoring service for Sealos, designed to track and analyze container image pulling issues with Prometheus metrics integration.
|
||||
|
||||
## Overview
|
||||
|
||||
Image Monitor Service acts as a monitoring layer for Kubernetes cluster image pull operations, providing:
|
||||
|
||||
- **Real-time Pod Monitoring**: Continuously watches Pod events using Kubernetes Informer mechanism
|
||||
- **Image Pull Failure Detection**: Automatically detects and classifies various image pull failures
|
||||
- **Slow Pull Alerts**: Tracks and alerts on slow image pull operations (>3 minutes)
|
||||
- **Prometheus Metrics**: Exports detailed metrics for monitoring and alerting
|
||||
- **Intelligent Classification**: Categorizes failures by root cause (network errors, auth issues, image not found, etc.)
|
||||
- **Graceful Shutdown**: Proper signal handling for zero-downtime deployments
|
||||
|
||||
## Architecture
|
||||
|
||||
```
|
||||
┌─────────────────┐
|
||||
│ Kubernetes │
|
||||
│ API Server │
|
||||
└────────┬────────┘
|
||||
│ Watch Pods
|
||||
▼
|
||||
┌──────────────────┐ ┌──────────────────┐
|
||||
│ Image Monitor │────▶│ Prometheus │
|
||||
│ Service │ │ Metrics │
|
||||
└──────────────────┘ └──────────────────┘
|
||||
│
|
||||
▼
|
||||
┌──────────────┐
|
||||
│ Failure │
|
||||
│ Analyzer │
|
||||
│ - Network │
|
||||
│ - Auth │
|
||||
│ - Not Found │
|
||||
│ - Slow Pull │
|
||||
└──────────────┘
|
||||
```
|
||||
|
||||
### Key Components
|
||||
|
||||
- **Pod Informer**: Watches all Pod resources across the cluster, tracking image pull status changes
|
||||
- **Failure Analyzer** (`analyzer.go`): Analyzes container status and identifies image pull errors
|
||||
- **Failure Classifier** (`classifier.go`): Categorizes failures into specific types using regex pattern matching
|
||||
- **Slow Pull Tracker** (`slow_pull.go`): Monitors pulling duration and triggers alerts for slow operations
|
||||
- **Metrics Exporter** (`metrics.go`): Exposes Prometheus metrics at `:8080/metrics`
|
||||
|
||||
## Features
|
||||
|
||||
- **Comprehensive Failure Classification**: Automatically identifies and categorizes:
|
||||
- Image not found (404, manifest unknown)
|
||||
- Proxy connection errors
|
||||
- Authentication/authorization failures
|
||||
- TLS handshake failures
|
||||
- I/O timeout issues
|
||||
- Connection refused errors
|
||||
- Network request failures
|
||||
- Back-off states
|
||||
|
||||
- **Slow Pull Detection**:
|
||||
- Tracks images in "ContainerCreating" state
|
||||
- Alerts when pulling exceeds 3-minute threshold
|
||||
- Clears alert when image finally succeeds or fails
|
||||
|
||||
- **Smart State Management**:
|
||||
- Preserves specific failure reasons over generic back-off states
|
||||
- Cleans up metrics when Pods are deleted
|
||||
- Handles both init containers and regular containers
|
||||
|
||||
- **Public Registry Focus**: Monitors only public registry images to avoid exposing private infrastructure
|
||||
|
||||
## Prerequisites
|
||||
|
||||
- Kubernetes cluster (v1.20+)
|
||||
- In-cluster deployment with appropriate RBAC permissions
|
||||
- Prometheus for metrics collection (optional but recommended)
|
||||
- Go 1.24+ (for development)
|
||||
|
||||
## Metrics
|
||||
|
||||
The service exposes two main Prometheus metrics:
|
||||
|
||||
### `image_pull_failure`
|
||||
Tracks active image pull failures with detailed labels.
|
||||
|
||||
**Type**: Gauge
|
||||
**Labels**:
|
||||
- `namespace`: Pod namespace
|
||||
- `pod`: Pod name
|
||||
- `node`: Node where pull is failing
|
||||
- `registry`: Container registry (e.g., docker.io, ghcr.io)
|
||||
- `image`: Full image reference
|
||||
- `reason`: Classified failure reason
|
||||
|
||||
**Example**:
|
||||
```promql
|
||||
image_pull_failure{
|
||||
namespace="default",
|
||||
pod="my-app-xyz",
|
||||
node="node-1",
|
||||
registry="docker.io",
|
||||
image="nginx:latest",
|
||||
reason="image_not_found"
|
||||
} 1
|
||||
```
|
||||
|
||||
### `image_pull_slow_alert`
|
||||
Tracks slow image pull operations (>3 minutes).
|
||||
|
||||
**Type**: Gauge
|
||||
**Labels**:
|
||||
- `namespace`: Pod namespace
|
||||
- `pod`: Pod name
|
||||
- `container`: Container name
|
||||
- `image`: Full image reference
|
||||
|
||||
**Example**:
|
||||
```promql
|
||||
image_pull_slow_alert{
|
||||
namespace="default",
|
||||
pod="my-app-xyz",
|
||||
container="app",
|
||||
image="docker.io/myimage:v1.0"
|
||||
} 1
|
||||
```
|
||||
@@ -0,0 +1,55 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
)
|
||||
|
||||
var (
|
||||
podFailures sync.Map // key namespace/pod -> *podInfo
|
||||
clientset *kubernetes.Clientset
|
||||
)
|
||||
|
||||
func checkImagePullError(
|
||||
statuses []corev1.ContainerStatus,
|
||||
nodeName string,
|
||||
pod *corev1.Pod,
|
||||
reasons map[string]failureInfo,
|
||||
) {
|
||||
for _, cs := range statuses {
|
||||
if cs.State.Waiting == nil ||
|
||||
!isImagePullFailureReason(cs.State.Waiting.Reason) ||
|
||||
!isPublicRegistry(cs.Image) {
|
||||
continue
|
||||
}
|
||||
|
||||
classified := classifyFailureReason(
|
||||
cs.State.Waiting.Reason,
|
||||
cs.State.Waiting.Message,
|
||||
)
|
||||
registry := parseRegistry(cs.Image)
|
||||
// Use container name as key
|
||||
reasons[cs.Name] = failureInfo{
|
||||
registry: registry,
|
||||
nodeName: nodeName, // Record current node
|
||||
image: cs.Image, // Record image info
|
||||
reason: classified, // Record failure reason
|
||||
}
|
||||
|
||||
// If in failure state, clean up corresponding slow pull state
|
||||
key := fmt.Sprintf("%s/%s/%s", pod.Namespace, pod.Name, cs.Name)
|
||||
cleanupSlowPull(key)
|
||||
}
|
||||
}
|
||||
|
||||
func analyzePodImagePullErrors(nodeName string, pod *corev1.Pod) map[string]failureInfo {
|
||||
reasons := make(map[string]failureInfo)
|
||||
|
||||
checkImagePullError(pod.Status.InitContainerStatuses, nodeName, pod, reasons)
|
||||
checkImagePullError(pod.Status.ContainerStatuses, nodeName, pod, reasons)
|
||||
|
||||
return reasons
|
||||
}
|
||||
@@ -0,0 +1,102 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"log"
|
||||
"regexp"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// Predefined regular expressions for different failure reasons, used to classify error messages
|
||||
var (
|
||||
reImageNotFound = regexp.MustCompile(
|
||||
`(?i)not found|NotFound|manifest unknown|repository does not exist`,
|
||||
)
|
||||
reProxyError = regexp.MustCompile(`(?i)proxyconnect|proxy error`)
|
||||
reUnauthorized = regexp.MustCompile(
|
||||
`(?i)unauthorized|authentication require|failed to authorize|authorization failed`,
|
||||
)
|
||||
reTLS = regexp.MustCompile(`(?i)tls handshake|failed to verify certificate`)
|
||||
reIOTimeout = regexp.MustCompile(`(?i)i/o timeout`)
|
||||
reConnectionRefused = regexp.MustCompile(`(?i)connection refused`)
|
||||
reNetworkError = regexp.MustCompile(`(?i)failed to do request`)
|
||||
)
|
||||
|
||||
// isBackOffPullingImage checks if the state is back-off pulling image
|
||||
func isBackOffPullingImage(reason, message string) bool {
|
||||
if strings.ToLower(reason) == "imagepullbackoff" {
|
||||
return true
|
||||
}
|
||||
|
||||
if strings.Contains(strings.ToLower(message), "back-off pulling image") {
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func isImagePullFailureReason(reason string) bool {
|
||||
switch reason {
|
||||
case "ErrImagePull", "ImagePullBackOff", "Cancelled", "RegistryUnavailable":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func isImagePullSlowReason(reason string) bool {
|
||||
switch reason {
|
||||
case "ContainerCreating":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func classifyFailureReason(r, message string) reason {
|
||||
lowMsg := strings.ToLower(message)
|
||||
switch strings.ToLower(r) {
|
||||
case "errimagepull", "imagepullbackoff":
|
||||
if reImageNotFound.MatchString(lowMsg) {
|
||||
return ReasonImageNotFound
|
||||
}
|
||||
|
||||
if reProxyError.MatchString(lowMsg) {
|
||||
return ReasonProxyError
|
||||
}
|
||||
|
||||
if reUnauthorized.MatchString(lowMsg) {
|
||||
return ReasonUnauthorized
|
||||
}
|
||||
|
||||
if reTLS.MatchString(lowMsg) {
|
||||
return ReasonTLSHandshake
|
||||
}
|
||||
|
||||
if reIOTimeout.MatchString(lowMsg) {
|
||||
return ReasonIOTimeout
|
||||
}
|
||||
|
||||
if reConnectionRefused.MatchString(lowMsg) {
|
||||
return ReasonConnectionRefused
|
||||
}
|
||||
|
||||
if reNetworkError.MatchString(lowMsg) {
|
||||
return ReasonNetworkError
|
||||
}
|
||||
|
||||
if strings.HasPrefix(lowMsg, "back-off pulling image") {
|
||||
return ReasonBackOff
|
||||
}
|
||||
|
||||
log.Printf("[Classify] Unknown error classification reason=%s message=%s", r, message)
|
||||
|
||||
return ReasonUnknown
|
||||
default:
|
||||
return strings.ToLower(r)
|
||||
}
|
||||
}
|
||||
|
||||
// isSpecificReason determines if it's a specific failure reason (not back_off_pulling_image)
|
||||
func isSpecificReason(reason string) bool {
|
||||
return reason != ReasonBackOff && reason != ReasonUnknown
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
module github.com/labring/sealos/service/imagemonitor
|
||||
|
||||
go 1.24.0
|
||||
|
||||
require (
|
||||
github.com/prometheus/client_golang v1.23.2
|
||||
k8s.io/api v0.34.1
|
||||
k8s.io/apimachinery v0.34.1
|
||||
k8s.io/client-go v12.0.0+incompatible
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/beorn7/perks v1.0.1 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
|
||||
github.com/fxamacker/cbor/v2 v2.9.0 // indirect
|
||||
github.com/go-logr/logr v1.4.2 // indirect
|
||||
github.com/gogo/protobuf v1.3.2 // indirect
|
||||
github.com/golang/protobuf v1.5.4 // indirect
|
||||
github.com/google/go-cmp v0.7.0 // indirect
|
||||
github.com/google/gofuzz v1.2.0 // indirect
|
||||
github.com/googleapis/gnostic v0.0.0-20170729233727-0c5108395e2d // indirect
|
||||
github.com/json-iterator/go v1.1.12 // indirect
|
||||
github.com/kr/text v0.2.0 // indirect
|
||||
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
|
||||
github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
|
||||
github.com/prometheus/client_model v0.6.2 // indirect
|
||||
github.com/prometheus/common v0.66.1 // indirect
|
||||
github.com/prometheus/procfs v0.16.1 // indirect
|
||||
github.com/rogpeppe/go-internal v1.13.1 // indirect
|
||||
github.com/spf13/pflag v1.0.6-0.20210604193023-d5e0c0615ace // indirect
|
||||
github.com/x448/float16 v0.8.4 // indirect
|
||||
go.yaml.in/yaml/v2 v2.4.2 // indirect
|
||||
go.yaml.in/yaml/v3 v3.0.4 // indirect
|
||||
golang.org/x/crypto v0.41.0 // indirect
|
||||
golang.org/x/net v0.43.0 // indirect
|
||||
golang.org/x/oauth2 v0.30.0 // indirect
|
||||
golang.org/x/sys v0.35.0 // indirect
|
||||
golang.org/x/term v0.34.0 // indirect
|
||||
golang.org/x/text v0.28.0 // indirect
|
||||
golang.org/x/time v0.9.0 // indirect
|
||||
google.golang.org/protobuf v1.36.8 // indirect
|
||||
gopkg.in/inf.v0 v0.9.1 // indirect
|
||||
gopkg.in/yaml.v2 v2.2.1 // indirect
|
||||
k8s.io/klog v0.3.1 // indirect
|
||||
k8s.io/klog/v2 v2.130.1 // indirect
|
||||
k8s.io/utils v0.0.0-20250604170112-4c0f3b243397 // indirect
|
||||
sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 // indirect
|
||||
sigs.k8s.io/structured-merge-diff/v4 v4.4.2 // indirect
|
||||
sigs.k8s.io/yaml v1.6.0 // indirect
|
||||
)
|
||||
@@ -0,0 +1,265 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
)
|
||||
|
||||
func onPodAddOrUpdate(obj any) {
|
||||
pod, ok := obj.(*corev1.Pod)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
currentNodeName := getNodeName(pod)
|
||||
if currentNodeName == "" {
|
||||
// not schedule to any node, skip
|
||||
return
|
||||
}
|
||||
|
||||
log.Printf(
|
||||
"[onPodAddOrUpdate] phase=%s uid=%s node=%s namespace=%s pod=%s containers=%d",
|
||||
pod.Status.Phase,
|
||||
string(pod.UID),
|
||||
currentNodeName,
|
||||
pod.Namespace,
|
||||
pod.Name,
|
||||
len(pod.Status.ContainerStatuses),
|
||||
)
|
||||
|
||||
// Iterate through InitContainerStatuses + ContainerStatuses
|
||||
for _, cs := range pod.Status.InitContainerStatuses {
|
||||
checkSlowPull(pod.Namespace, pod.Name, cs, cs.Image)
|
||||
}
|
||||
|
||||
for _, cs := range pod.Status.ContainerStatuses {
|
||||
checkSlowPull(pod.Namespace, pod.Name, cs, cs.Image)
|
||||
}
|
||||
|
||||
reasons := analyzePodImagePullErrors(currentNodeName, pod)
|
||||
|
||||
podKey := fmt.Sprintf("%s/%s", pod.Namespace, pod.Name)
|
||||
|
||||
piVal, _ := podFailures.LoadOrStore(podKey, &podInfo{
|
||||
reasons: make(map[string]failureInfo),
|
||||
podName: pod.Name,
|
||||
namespace: pod.Namespace,
|
||||
})
|
||||
|
||||
pi, ok := piVal.(*podInfo)
|
||||
if !ok {
|
||||
log.Printf("[onPodAddOrUpdate] Unable to parse added object type: %T", piVal)
|
||||
return
|
||||
}
|
||||
|
||||
pi.mu.Lock()
|
||||
defer pi.mu.Unlock()
|
||||
|
||||
pi.namespace = pod.Namespace
|
||||
pi.podName = pod.Name
|
||||
|
||||
updateReasons(pi, reasons)
|
||||
}
|
||||
|
||||
func onPodDelete(obj any) {
|
||||
pod, ok := obj.(*corev1.Pod)
|
||||
if !ok {
|
||||
tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
|
||||
if !ok {
|
||||
log.Printf("[onPodDelete] Unable to parse deleted object type: %T", obj)
|
||||
return
|
||||
}
|
||||
|
||||
pod, ok = tombstone.Obj.(*corev1.Pod)
|
||||
if !ok {
|
||||
log.Printf("[onPodDelete] Unable to convert tombstone object: %T", tombstone.Obj)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
log.Printf("[onPodDelete] namespace=%s pod=%s", pod.Namespace, pod.Name)
|
||||
|
||||
key := fmt.Sprintf("%s/%s", pod.Namespace, pod.Name)
|
||||
|
||||
reasonsVal, loaded := podFailures.LoadAndDelete(key)
|
||||
if loaded {
|
||||
pi, ok := reasonsVal.(*podInfo)
|
||||
if !ok {
|
||||
log.Printf("[onPodDelete] Unable to parse deleted object type: %T", reasonsVal)
|
||||
return
|
||||
}
|
||||
|
||||
pi.mu.Lock()
|
||||
defer pi.mu.Unlock()
|
||||
|
||||
log.Printf(
|
||||
"[onPodDelete] namespace=%s pod=%s cleanup %d reasons",
|
||||
pi.namespace,
|
||||
pi.podName,
|
||||
len(pi.reasons),
|
||||
)
|
||||
|
||||
// Use stored node information for Dec operation to ensure execution on correct node
|
||||
for containerName, info := range pi.reasons {
|
||||
log.Printf(
|
||||
"[onPodDelete] Dec gauge: namespace=%s pod=%s container=%s node=%s registry=%s image=%s reason=%s",
|
||||
pi.namespace,
|
||||
pi.podName,
|
||||
containerName,
|
||||
info.nodeName,
|
||||
info.registry,
|
||||
info.image,
|
||||
info.reason,
|
||||
)
|
||||
imagePullFailureGauge.DeleteLabelValues(
|
||||
pi.namespace,
|
||||
pi.podName,
|
||||
info.nodeName,
|
||||
info.registry,
|
||||
info.image,
|
||||
info.reason,
|
||||
)
|
||||
}
|
||||
} else {
|
||||
log.Printf("[onPodDelete] pod %s not found in podFailures", key)
|
||||
}
|
||||
|
||||
// Clean up slow pull related state
|
||||
prefix := key + "/"
|
||||
slowPullTimers.Range(newCleanupSlowPullWithPrefixFunc(prefix))
|
||||
slowPullTracking.Range(newCleanupSlowPullWithPrefixFunc(prefix))
|
||||
}
|
||||
|
||||
func updateReasons(
|
||||
pi *podInfo,
|
||||
reasons map[string]failureInfo,
|
||||
) {
|
||||
// Remove old reasons - use stored node information
|
||||
for containerName, oldInfo := range pi.reasons {
|
||||
if _, found := reasons[containerName]; !found {
|
||||
log.Printf(
|
||||
"[UpdateReasons] Dec gauge: namespace=%s pod=%s container=%s node=%s registry=%s image=%s reason=%s",
|
||||
pi.namespace,
|
||||
pi.podName,
|
||||
containerName,
|
||||
oldInfo.nodeName,
|
||||
oldInfo.registry,
|
||||
oldInfo.image,
|
||||
oldInfo.reason,
|
||||
)
|
||||
imagePullFailureGauge.DeleteLabelValues(
|
||||
pi.namespace,
|
||||
pi.podName,
|
||||
oldInfo.nodeName,
|
||||
oldInfo.registry,
|
||||
oldInfo.image,
|
||||
oldInfo.reason,
|
||||
)
|
||||
delete(pi.reasons, containerName)
|
||||
}
|
||||
}
|
||||
|
||||
// Add new reasons
|
||||
for containerName, info := range reasons {
|
||||
oldInfo, found := pi.reasons[containerName]
|
||||
if found {
|
||||
// Check if we need to preserve the existing specific reason
|
||||
finalReason := info.reason
|
||||
|
||||
// If the new reason is back_off_pulling_image, and other info hasn't changed, and there was a specific reason before, preserve the specific reason
|
||||
if info.reason == ReasonBackOff &&
|
||||
oldInfo.nodeName == info.nodeName &&
|
||||
oldInfo.image == info.image &&
|
||||
oldInfo.registry == info.registry &&
|
||||
isSpecificReason(oldInfo.reason) {
|
||||
finalReason = oldInfo.reason
|
||||
log.Printf(
|
||||
"[UpdateReasons] Preserving specific reason for %s/%s container=%s: keeping '%s' instead of 'back_off_pulling_image'",
|
||||
pi.namespace,
|
||||
pi.podName,
|
||||
containerName,
|
||||
oldInfo.reason,
|
||||
)
|
||||
}
|
||||
|
||||
// Check if there are changes (node, failure reason, image, etc.)
|
||||
if oldInfo.nodeName != info.nodeName || oldInfo.reason != finalReason ||
|
||||
oldInfo.image != info.image {
|
||||
log.Printf(
|
||||
"[UpdateReasons] Info changed for %s/%s container=%s: node=%s->%s, reason=%s->%s, image=%s->%s",
|
||||
pi.namespace,
|
||||
pi.podName,
|
||||
containerName,
|
||||
oldInfo.nodeName,
|
||||
info.nodeName,
|
||||
oldInfo.reason,
|
||||
finalReason,
|
||||
oldInfo.image,
|
||||
info.image,
|
||||
)
|
||||
|
||||
// Dec on old information
|
||||
imagePullFailureGauge.DeleteLabelValues(
|
||||
pi.namespace,
|
||||
pi.podName,
|
||||
oldInfo.nodeName,
|
||||
oldInfo.registry,
|
||||
oldInfo.image,
|
||||
oldInfo.reason,
|
||||
)
|
||||
// Inc on new information
|
||||
imagePullFailureGauge.WithLabelValues(pi.namespace, pi.podName, info.nodeName, info.registry, info.image, finalReason).
|
||||
Set(1)
|
||||
|
||||
// Update stored information using the final determined reason
|
||||
info.reason = finalReason
|
||||
pi.reasons[containerName] = info
|
||||
} else if oldInfo.reason != finalReason {
|
||||
// Case where only the reason has changed
|
||||
log.Printf(
|
||||
"[UpdateReasons] Only reason changed for %s/%s container=%s: %s->%s",
|
||||
pi.namespace,
|
||||
pi.podName,
|
||||
containerName,
|
||||
oldInfo.reason,
|
||||
finalReason,
|
||||
)
|
||||
|
||||
imagePullFailureGauge.DeleteLabelValues(
|
||||
pi.namespace,
|
||||
pi.podName,
|
||||
oldInfo.nodeName,
|
||||
oldInfo.registry,
|
||||
oldInfo.image,
|
||||
oldInfo.reason,
|
||||
)
|
||||
imagePullFailureGauge.WithLabelValues(pi.namespace, pi.podName, info.nodeName, info.registry, info.image, finalReason).
|
||||
Set(1)
|
||||
|
||||
info.reason = finalReason
|
||||
pi.reasons[containerName] = info
|
||||
}
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
// Brand new failed container
|
||||
log.Printf(
|
||||
"[UpdateReasons] Inc gauge: namespace=%s pod=%s container=%s node=%s registry=%s image=%s reason=%s",
|
||||
pi.namespace,
|
||||
pi.podName,
|
||||
containerName,
|
||||
info.nodeName,
|
||||
info.registry,
|
||||
info.image,
|
||||
info.reason,
|
||||
)
|
||||
imagePullFailureGauge.WithLabelValues(pi.namespace, pi.podName, info.nodeName, info.registry, info.image, info.reason).
|
||||
Set(1)
|
||||
|
||||
pi.reasons[containerName] = info
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||
"k8s.io/client-go/informers"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
"k8s.io/client-go/rest"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
)
|
||||
|
||||
func main() {
|
||||
// Register Prometheus metrics
|
||||
prometheus.MustRegister(imagePullFailureGauge)
|
||||
prometheus.MustRegister(imagePullSlowAlertGauge)
|
||||
|
||||
// Create in-cluster config
|
||||
config, err := rest.InClusterConfig()
|
||||
if err != nil {
|
||||
log.Fatalf("Error creating in-cluster config: %v", err)
|
||||
}
|
||||
|
||||
clientset, err = kubernetes.NewForConfig(config)
|
||||
if err != nil {
|
||||
log.Fatalf("Error creating Kubernetes clientset: %v", err)
|
||||
}
|
||||
|
||||
// Create informer
|
||||
factory := informers.NewSharedInformerFactory(clientset, 0)
|
||||
podInformer := factory.Core().V1().Pods().Informer()
|
||||
_, _ = podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: onPodAddOrUpdate,
|
||||
UpdateFunc: func(_, obj any) { onPodAddOrUpdate(obj) },
|
||||
DeleteFunc: onPodDelete,
|
||||
})
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
go podInformer.Run(ctx.Done())
|
||||
|
||||
if !cache.WaitForCacheSync(ctx.Done(), podInformer.HasSynced) {
|
||||
//nolint:gocritic
|
||||
log.Fatalf("Timed out waiting for caches to sync")
|
||||
}
|
||||
|
||||
// HTTP server and graceful shutdown
|
||||
srv := &http.Server{
|
||||
Addr: ":8080",
|
||||
Handler: promhttp.Handler(),
|
||||
ReadHeaderTimeout: 5 * time.Second,
|
||||
}
|
||||
go func() {
|
||||
log.Println("Starting metrics server at :8080")
|
||||
|
||||
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
log.Fatalf("Metrics server error: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
// Capture signals
|
||||
sigCh := make(chan os.Signal, 1)
|
||||
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
|
||||
<-sigCh
|
||||
log.Println("Shutdown signal received, exiting...")
|
||||
|
||||
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer shutdownCancel()
|
||||
|
||||
if err := srv.Shutdown(shutdownCtx); err != nil {
|
||||
log.Printf("Error during server shutdown: %v", err)
|
||||
}
|
||||
|
||||
log.Println("Server gracefully stopped")
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
)
|
||||
|
||||
// Define Prometheus metrics
|
||||
var (
|
||||
//nolint:promlinter
|
||||
imagePullFailureGauge = prometheus.NewGaugeVec(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "k8s_pod_image_pull_failure_total",
|
||||
Help: "Number of pods with image pull failures categorized by exported_namespace, exported_pod, node, image and reason",
|
||||
},
|
||||
[]string{"exported_namespace", "exported_pod", "node", "registry", "image", "reason"},
|
||||
)
|
||||
// Changed to Gauge type, allowing Inc and Dec operations
|
||||
//nolint:promlinter
|
||||
imagePullSlowAlertGauge = prometheus.NewGaugeVec(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "k8s_pod_image_pull_slow_total",
|
||||
Help: "Number of pods with slow image pull (>=5m), by exported_namespace, exported_pod, node, registry and image",
|
||||
},
|
||||
[]string{"exported_namespace", "exported_pod", "node", "registry", "image"},
|
||||
)
|
||||
)
|
||||
@@ -0,0 +1,165 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
)
|
||||
|
||||
// slowPullTimers stores image pull timers
|
||||
var slowPullTimers sync.Map // key:string -> *time.Timer
|
||||
|
||||
// slowPullTracking tracks current slow pull state
|
||||
var slowPullTracking sync.Map // key: namespace/pod/container -> slowPullInfo
|
||||
|
||||
func newCheckSlowPullHandler(
|
||||
slowPullTimerKey, ns, podName string,
|
||||
cs corev1.ContainerStatus,
|
||||
image string,
|
||||
) func() {
|
||||
return func() {
|
||||
slowPullTimers.Delete(slowPullTimerKey)
|
||||
|
||||
p2, err := clientset.CoreV1().
|
||||
Pods(ns).
|
||||
Get(context.Background(), podName, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[SlowPull] Failed to get Pod %s/%s: %v", ns, podName, err)
|
||||
return
|
||||
}
|
||||
|
||||
nodeName := getNodeName(p2)
|
||||
if nodeName == "" {
|
||||
// not schedule to any node, skip
|
||||
return
|
||||
}
|
||||
|
||||
for _, newCs := range p2.Status.ContainerStatuses {
|
||||
if newCs.Name != cs.Name {
|
||||
continue
|
||||
}
|
||||
|
||||
// Check if container is still waiting and not in failure state
|
||||
if newCs.ContainerID != "" ||
|
||||
newCs.State.Waiting == nil ||
|
||||
!isImagePullSlowReason(newCs.State.Waiting.Reason) ||
|
||||
isBackOffPullingImage(
|
||||
newCs.State.Waiting.Reason,
|
||||
newCs.State.Waiting.Message,
|
||||
) {
|
||||
break
|
||||
}
|
||||
|
||||
registry := parseRegistry(image)
|
||||
|
||||
// Record slow pull state
|
||||
slowPullInfo := slowPullInfo{
|
||||
namespace: ns,
|
||||
podName: podName,
|
||||
nodeName: nodeName,
|
||||
registry: registry,
|
||||
image: image,
|
||||
}
|
||||
slowPullTracking.Store(slowPullTimerKey, slowPullInfo)
|
||||
|
||||
// Increment slow pull metric
|
||||
imagePullSlowAlertGauge.WithLabelValues(ns, podName, nodeName, registry, image).
|
||||
Set(1)
|
||||
|
||||
log.Printf(
|
||||
"[SlowPullAlert] %s/%s container=%s node=%s registry=%s image=%s",
|
||||
ns,
|
||||
podName,
|
||||
cs.Name,
|
||||
nodeName,
|
||||
registry,
|
||||
image,
|
||||
)
|
||||
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func checkSlowPull(ns, podName string, cs corev1.ContainerStatus, image string) {
|
||||
slowPullTimerKey := fmt.Sprintf("%s/%s/%s", ns, podName, cs.Name)
|
||||
|
||||
// Check if in image pull slow state
|
||||
if cs.ContainerID != "" ||
|
||||
cs.State.Waiting == nil ||
|
||||
!isImagePullSlowReason(cs.State.Waiting.Reason) ||
|
||||
isBackOffPullingImage(cs.State.Waiting.Reason, cs.State.Waiting.Message) {
|
||||
cleanupSlowPull(slowPullTimerKey)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// Check if a timer is already running
|
||||
if _, exists := slowPullTimers.Load(slowPullTimerKey); exists {
|
||||
// Timer already exists, no need to create duplicate
|
||||
return
|
||||
}
|
||||
|
||||
timer := time.AfterFunc(
|
||||
5*time.Minute,
|
||||
newCheckSlowPullHandler(slowPullTimerKey, ns, podName, cs, image),
|
||||
)
|
||||
|
||||
_, loaded := slowPullTimers.LoadOrStore(slowPullTimerKey, timer)
|
||||
if loaded {
|
||||
timer.Stop()
|
||||
log.Printf(
|
||||
"[SlowPull] Timer already exists for %s/%s container=%s, stopped duplicate timer",
|
||||
ns,
|
||||
podName,
|
||||
cs.Name,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
func newCleanupSlowPullWithPrefixFunc(prefix string) func(k, v any) bool {
|
||||
return func(k, v any) bool {
|
||||
if sk, ok := k.(string); ok && strings.HasPrefix(sk, prefix) {
|
||||
cleanupSlowPull(sk)
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// cleanupSlowPull cleans up slow pull state
|
||||
func cleanupSlowPull(slowPullTimerKey string) {
|
||||
if val, exists := slowPullTracking.LoadAndDelete(slowPullTimerKey); exists {
|
||||
if info, ok := val.(slowPullInfo); ok {
|
||||
imagePullSlowAlertGauge.DeleteLabelValues(
|
||||
info.namespace,
|
||||
info.podName,
|
||||
info.nodeName,
|
||||
info.registry,
|
||||
info.image,
|
||||
)
|
||||
log.Printf(
|
||||
"[SlowPullCleanup] Dec slow pull gauge: namespace=%s pod=%s node=%s registry=%s image=%s",
|
||||
info.namespace,
|
||||
info.podName,
|
||||
info.nodeName,
|
||||
info.registry,
|
||||
info.image,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// Also clean up timer
|
||||
if val, exists := slowPullTimers.LoadAndDelete(slowPullTimerKey); exists {
|
||||
if t, ok := val.(*time.Timer); ok {
|
||||
t.Stop()
|
||||
log.Printf("[SlowPullCleanup] Stopped timer for key: %s", slowPullTimerKey)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"sync"
|
||||
)
|
||||
|
||||
type reason = string
|
||||
|
||||
const (
|
||||
ReasonImageNotFound reason = "image_not_found"
|
||||
ReasonProxyError reason = "proxy_error"
|
||||
ReasonUnauthorized reason = "unauthorized"
|
||||
ReasonTLSHandshake reason = "tls_handshake_error"
|
||||
ReasonIOTimeout reason = "io_timeout"
|
||||
ReasonConnectionRefused reason = "connection_refused"
|
||||
ReasonNetworkError reason = "network_error"
|
||||
ReasonBackOff reason = "back_off_pulling_image"
|
||||
ReasonUnknown reason = "unknown"
|
||||
)
|
||||
|
||||
// podInfo contains failure reasons, node information and lock
|
||||
type podInfo struct {
|
||||
mu sync.Mutex
|
||||
reasons map[string]failureInfo // key: container name, value: failureInfo with node info
|
||||
namespace string
|
||||
podName string
|
||||
}
|
||||
|
||||
// failureInfo now includes node and image information to ensure Dec operations execute on the correct node
|
||||
type failureInfo struct {
|
||||
registry string // Image registry
|
||||
nodeName string // Node information
|
||||
image string // Image information
|
||||
reason reason // Failure reason
|
||||
}
|
||||
|
||||
// slowPullInfo records slow pull information
|
||||
type slowPullInfo struct {
|
||||
namespace string
|
||||
podName string
|
||||
nodeName string
|
||||
registry string
|
||||
image string
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
)
|
||||
|
||||
func getNodeName(pod *corev1.Pod) string {
|
||||
if pod.Spec.NodeName != "" {
|
||||
return pod.Spec.NodeName
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func isPublicRegistry(image string) bool {
|
||||
if !strings.Contains(image, "/") {
|
||||
return true
|
||||
}
|
||||
|
||||
return strings.HasPrefix(image, "docker.io/") ||
|
||||
strings.HasPrefix(image, "gcr.io/") ||
|
||||
strings.HasPrefix(image, "ghcr.io/") ||
|
||||
strings.HasPrefix(image, "k8s.gcr.io/") ||
|
||||
strings.HasPrefix(image, "quay.io/") ||
|
||||
strings.HasPrefix(image, "registry.k8s.io/") ||
|
||||
(strings.HasPrefix(image, "registry.") && strings.Contains(image, ".aliyuncs.com/")) ||
|
||||
(strings.HasPrefix(image, "hub.") && strings.Contains(image, ".sealos.run/")) ||
|
||||
strings.HasPrefix(image, "sealos.hub") ||
|
||||
strings.Contains(image, ".cr.aliyuncs.com/")
|
||||
}
|
||||
|
||||
func parseRegistry(image string) string {
|
||||
if image == "" {
|
||||
return "unknown"
|
||||
}
|
||||
|
||||
parts := strings.Split(image, "/")
|
||||
if len(parts) > 1 && strings.Contains(parts[0], ".") {
|
||||
return parts[0]
|
||||
}
|
||||
|
||||
return "docker.io"
|
||||
}
|
||||
Reference in New Issue
Block a user