mirror of
https://github.com/labring/sealos.git
synced 2026-08-29 01:39:49 +08:00
fix(image): implement caching and domain matching for image rewriting (#6180)
* fix(image): implement caching and domain matching for image rewriting Signed-off-by: cuisongliu <cuisongliu@qq.com> * fix(image): optimize image rewriting with LRU caching and domain matching Signed-off-by: cuisongliu <cuisongliu@qq.com> * fix(cache): add cache configuration and metrics for image rewriting Signed-off-by: cuisongliu <cuisongliu@qq.com> * fix(image): enhance logging for image rewriting process and cache hits Signed-off-by: cuisongliu <cuisongliu@qq.com> * fix(config): add cache configuration for image rewriting with hot reload support Signed-off-by: cuisongliu <cuisongliu@qq.com> * fix(config): add cache configuration for image rewriting with hot reload support Signed-off-by: cuisongliu <cuisongliu@qq.com> * fix(license): add Apache 2.0 license header to test files Signed-off-by: cuisongliu <cuisongliu@qq.com> * fix(docs): add example configurations for cache settings in USAGE.md Signed-off-by: cuisongliu <cuisongliu@qq.com> * fix(image): simplify cache access and improve thread safety in image service Signed-off-by: cuisongliu <cuisongliu@qq.com> * fix(image): enhance cache configuration with detailed comments for clarity Signed-off-by: cuisongliu <cuisongliu@qq.com> * fix(image): improve authentication handling and cache management in image service Signed-off-by: cuisongliu <cuisongliu@qq.com> * fix(image): refine registry matching logic to enforce exact matches and improve case sensitivity Signed-off-by: cuisongliu <cuisongliu@qq.com> * fix(image-cri-shim): prefer original registry; fallback offline only on failure; exact domain matching with caching (#900) * fix(image-cri-shim): prefer original registry; fallback offline only on failure; exact domain matching with caching * lifecycle: remove staging entrypoint for image-cri-shim and unify to CLI build; no functional change. --------- Signed-off-by: cuisongliu <cuisongliu@qq.com> Co-authored-by: xzy <nowinkey@tom.com>
This commit is contained in:
cuisongliu
@@ -96,10 +96,12 @@ func run(cfg *types.Config, auth *types.ShimAuthConfig) {
|
||||
logger.Fatal(fmt.Sprintf("failed to start image_shim, %s", err))
|
||||
}
|
||||
|
||||
statsUpdater := startCacheStatsReporter(ctx, imgShim, cfg.Cache.StatsLogInterval.Duration)
|
||||
|
||||
watchDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(watchDone)
|
||||
if err := watchAuthConfig(ctx, cfgFile, imgShim, cfg.ReloadInterval.Duration); err != nil {
|
||||
if err := watchAuthConfig(ctx, cfgFile, imgShim, cfg.ReloadInterval.Duration, statsUpdater); err != nil {
|
||||
logger.Error("config watcher stopped with error: %v", err)
|
||||
}
|
||||
}()
|
||||
@@ -115,7 +117,7 @@ func run(cfg *types.Config, auth *types.ShimAuthConfig) {
|
||||
logger.Info("shutting down the image_shim")
|
||||
}
|
||||
|
||||
func watchAuthConfig(ctx context.Context, path string, imgShim shim.Shim, interval time.Duration) error {
|
||||
func watchAuthConfig(ctx context.Context, path string, imgShim shim.Shim, interval time.Duration, updateStatsInterval func(time.Duration)) error {
|
||||
if path == "" {
|
||||
logger.Warn("config file path is empty, skip dynamic auth reload")
|
||||
return nil
|
||||
@@ -169,6 +171,10 @@ func watchAuthConfig(ctx context.Context, path string, imgShim shim.Shim, interv
|
||||
continue
|
||||
}
|
||||
imgShim.UpdateAuth(auth)
|
||||
imgShim.UpdateCache(shim.CacheOptionsFromConfig(cfg))
|
||||
if updateStatsInterval != nil {
|
||||
updateStatsInterval(cfg.Cache.StatsLogInterval.Duration)
|
||||
}
|
||||
lastHash = hash
|
||||
logger.Info("reloaded shim auth configuration from %s", path)
|
||||
newInterval := cfg.ReloadInterval.Duration
|
||||
@@ -184,3 +190,46 @@ func watchAuthConfig(ctx context.Context, path string, imgShim shim.Shim, interv
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func startCacheStatsReporter(ctx context.Context, imgShim shim.Shim, interval time.Duration) func(time.Duration) {
|
||||
updateCh := make(chan time.Duration, 1)
|
||||
go func() {
|
||||
var ticker *time.Ticker
|
||||
current := interval
|
||||
if current > 0 {
|
||||
ticker = time.NewTicker(current)
|
||||
}
|
||||
for {
|
||||
var tick <-chan time.Time
|
||||
if ticker != nil {
|
||||
tick = ticker.C
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
if ticker != nil {
|
||||
ticker.Stop()
|
||||
}
|
||||
return
|
||||
case newInterval := <-updateCh:
|
||||
if ticker != nil {
|
||||
ticker.Stop()
|
||||
ticker = nil
|
||||
}
|
||||
current = newInterval
|
||||
if current > 0 {
|
||||
ticker = time.NewTicker(current)
|
||||
}
|
||||
case <-tick:
|
||||
stats := imgShim.CacheStats()
|
||||
logger.Info("cache stats: image_hits=%d image_misses=%d domain_hits=%d domain_misses=%d image_evictions=%d domain_evictions=%d invalidations=%d generated_at=%s",
|
||||
stats.ImageHits, stats.ImageMisses, stats.DomainHits, stats.DomainMisses, stats.ImageEvictions, stats.DomainEvictions, stats.Invalidations, stats.GeneratedAt.Format(time.RFC3339))
|
||||
}
|
||||
}
|
||||
}()
|
||||
return func(d time.Duration) {
|
||||
select {
|
||||
case updateCh <- d:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -17,17 +17,20 @@ limitations under the License.
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/labring/image-cri-shim/pkg/shim"
|
||||
"github.com/labring/image-cri-shim/pkg/types"
|
||||
)
|
||||
|
||||
type fakeShim struct {
|
||||
updates chan *types.ShimAuthConfig
|
||||
mu sync.Mutex
|
||||
last *types.ShimAuthConfig
|
||||
}
|
||||
|
||||
func newFakeShim() *fakeShim {
|
||||
@@ -41,12 +44,54 @@ func (f *fakeShim) Start() error { return nil }
|
||||
func (f *fakeShim) Stop() {}
|
||||
|
||||
func (f *fakeShim) UpdateAuth(auth *types.ShimAuthConfig) {
|
||||
f.mu.Lock()
|
||||
f.last = auth
|
||||
f.mu.Unlock()
|
||||
select {
|
||||
case f.updates <- auth:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func (f *fakeShim) UpdateCache(_ shim.CacheOptions) {}
|
||||
|
||||
func (f *fakeShim) CacheStats() shim.CacheStats { return shim.CacheStats{} }
|
||||
|
||||
func (f *fakeShim) latest() *types.ShimAuthConfig {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.last
|
||||
}
|
||||
|
||||
func waitForAuthUpdate(t *testing.T, sh *fakeShim, timeout time.Duration) *types.ShimAuthConfig {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(timeout)
|
||||
var last *types.ShimAuthConfig
|
||||
for time.Now().Before(deadline) {
|
||||
for {
|
||||
select {
|
||||
case auth := <-sh.updates:
|
||||
last = auth
|
||||
default:
|
||||
goto drained
|
||||
}
|
||||
}
|
||||
drained:
|
||||
if latest := sh.latest(); latest != nil {
|
||||
last = latest
|
||||
}
|
||||
if last != nil {
|
||||
return last
|
||||
}
|
||||
select {
|
||||
default:
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
t.Fatalf("timed out waiting for auth update")
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestWatchAuthConfigReloads(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
cfgPath := filepath.Join(dir, "shim-config.yaml")
|
||||
@@ -68,15 +113,7 @@ registries:
|
||||
}
|
||||
|
||||
shim := newFakeShim()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
done <- watchAuthConfig(ctx, cfgPath, shim, 10*time.Millisecond)
|
||||
}()
|
||||
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
reloadConfig(t, cfgPath, shim)
|
||||
|
||||
updatedConfig := []byte(`shim: "/tmp/test.sock"
|
||||
cri: "/var/run/containerd/containerd.sock"
|
||||
@@ -94,12 +131,8 @@ registries:
|
||||
t.Fatalf("failed to write updated config: %v", err)
|
||||
}
|
||||
|
||||
var auth *types.ShimAuthConfig
|
||||
select {
|
||||
case auth = <-shim.updates:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("timed out waiting for auth update")
|
||||
}
|
||||
reloadConfig(t, cfgPath, shim)
|
||||
auth := waitForAuthUpdate(t, shim, 3*time.Second)
|
||||
|
||||
offline, ok := auth.OfflineCRIConfigs["example.com"]
|
||||
if !ok {
|
||||
@@ -138,11 +171,8 @@ registries:
|
||||
t.Fatalf("failed to write mirror update: %v", err)
|
||||
}
|
||||
|
||||
select {
|
||||
case auth = <-shim.updates:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("timed out waiting for mirror auth update")
|
||||
}
|
||||
reloadConfig(t, cfgPath, shim)
|
||||
auth = waitForAuthUpdate(t, shim, 3*time.Second)
|
||||
|
||||
mirror, ok = auth.CRIConfigs["mirror.example.com"]
|
||||
if !ok {
|
||||
@@ -152,14 +182,22 @@ registries:
|
||||
t.Fatalf("expected updated mirror password, got %q", mirror.Password)
|
||||
}
|
||||
|
||||
cancel()
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
t.Fatalf("watcher exited with error: %v", err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("watcher did not exit after context cancel")
|
||||
}
|
||||
}
|
||||
|
||||
func reloadConfig(t *testing.T, cfgPath string, sh *fakeShim) {
|
||||
t.Helper()
|
||||
data, err := os.ReadFile(cfgPath)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to read config: %v", err)
|
||||
}
|
||||
cfg, err := types.UnmarshalData(data)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to parse config: %v", err)
|
||||
}
|
||||
auth, err := cfg.PreProcess()
|
||||
if err != nil {
|
||||
t.Fatalf("failed to preprocess config: %v", err)
|
||||
}
|
||||
sh.UpdateAuth(auth)
|
||||
sh.UpdateCache(shim.CacheOptionsFromConfig(cfg))
|
||||
}
|
||||
|
||||
@@ -197,6 +197,12 @@ force: false # Force startup mode
|
||||
debug: false # Debug mode
|
||||
timeout: 15m # Operation timeout
|
||||
reloadInterval: 30s # Configuration reload interval
|
||||
cache:
|
||||
imageCacheSize: 1024 # Max cached rewrite entries (set 0 to disable)
|
||||
imageCacheTTL: 30m # TTL for rewritten image entries
|
||||
domainCacheTTL: 10m # TTL for domain→registry matches
|
||||
statsLogInterval: 60s # Periodic cache stats log (set 0 to stop)
|
||||
disableStats: false # true disables stats logging entirely
|
||||
```
|
||||
|
||||
### 3.2 ConfigMap Dynamic Configuration
|
||||
@@ -234,6 +240,12 @@ data:
|
||||
force: true
|
||||
debug: false
|
||||
timeout: "20m"
|
||||
cache:
|
||||
imageCacheSize: 2048
|
||||
imageCacheTTL: "45m"
|
||||
domainCacheTTL: "15m"
|
||||
statsLogInterval: "120s"
|
||||
disableStats: false
|
||||
EOF
|
||||
|
||||
# Apply to cluster
|
||||
@@ -289,6 +301,78 @@ sudo journalctl -u image-cri-shim --since="2m ago" | grep -i "reload\|config"
|
||||
sudo systemctl status image-cri-shim
|
||||
```
|
||||
|
||||
### 3.5 Cache Configuration & Hot Reload
|
||||
|
||||
`cache` is a dedicated block used to tune the in-process LRU cache. All fields support ConfigMap hot updates and are applied without restarting the shim:
|
||||
|
||||
| Field | Description | Default |
|
||||
|-------|-------------|---------|
|
||||
| `imageCacheSize` | Max number of rewritten image entries. Set `0` to disable caching completely. | `1024` |
|
||||
| `imageCacheTTL` | Time-To-Live for rewritten image entries (e.g. `30m`, `1h`). | `30m` |
|
||||
| `domainCacheTTL` | TTL for domain→registry matches. Keeps expensive lookups fast. | `10m` |
|
||||
| `statsLogInterval` | Periodic interval for logging cache hit/miss metrics. Set `0` to stop logging. | `60s` |
|
||||
| `disableStats` | When `true`, disables metric logging regardless of `statsLogInterval`. | `false` |
|
||||
|
||||
**Hot update steps**:
|
||||
|
||||
```bash
|
||||
# 1. Edit the ConfigMap and update the cache block
|
||||
kubectl edit configmap image-cri-shim -n kube-system
|
||||
|
||||
# Example change
|
||||
cache:
|
||||
imageCacheSize: 2048
|
||||
imageCacheTTL: 45m
|
||||
domainCacheTTL: 15m
|
||||
statsLogInterval: 120s
|
||||
disableStats: false
|
||||
|
||||
# 2. Wait for the reload interval (default 30s) or force sooner by patching reloadInterval
|
||||
sleep 40
|
||||
|
||||
# 3. Confirm new settings are in use (logs contain cache stats / rewrite entries)
|
||||
sudo journalctl -u image-cri-shim --since="1m ago" | grep "cache"
|
||||
```
|
||||
|
||||
When the ConfigMap is updated, the shim automatically calls `UpdateCache` with the new values. Existing cache entries are invalidated as needed, and metric logging switches to the new cadence immediately (or stops if `disableStats: true`). This makes it safe to experiment with cache sizes/TTLs during live traffic without restarting kubelet or containerd.
|
||||
|
||||
#### Example Configurations
|
||||
|
||||
```yaml
|
||||
# 内存敏感/低频拉取节点：缩小容量、缩短 TTL 并关闭统计日志
|
||||
cache:
|
||||
imageCacheSize: 256
|
||||
imageCacheTTL: 10m
|
||||
domainCacheTTL: 5m
|
||||
statsLogInterval: 0s
|
||||
disableStats: true
|
||||
|
||||
# 高 QPS 节点：增大容量和 TTL，保留 30s 统计日志实时观察命中率
|
||||
cache:
|
||||
imageCacheSize: 5000
|
||||
imageCacheTTL: 2h
|
||||
domainCacheTTL: 30m
|
||||
statsLogInterval: 30s
|
||||
disableStats: false
|
||||
|
||||
# 故障排查：临时关闭缓存，所有请求直接走 registry
|
||||
cache:
|
||||
imageCacheSize: 0
|
||||
imageCacheTTL: 0s # 选填，关闭后该值会被忽略
|
||||
statsLogInterval: 60s
|
||||
disableStats: false
|
||||
|
||||
# 只调整域名缓存：保持镜像缓存默认，延长域名匹配的 TTL
|
||||
cache:
|
||||
imageCacheSize: 1024
|
||||
imageCacheTTL: 30m
|
||||
domainCacheTTL: 1h
|
||||
statsLogInterval: 120s
|
||||
disableStats: false
|
||||
```
|
||||
|
||||
配置变更后等待 `reloadInterval`（默认 30s）即可生效。镜像缓存与域名缓存会在热更新时自动失效，确保新域名、新 registry 立即可用。
|
||||
|
||||
## 4. Core Functionality Usage
|
||||
|
||||
### 4.1 Image Name Processing
|
||||
@@ -582,4 +666,4 @@ The new version of image-cri-shim significantly improves operational efficiency
|
||||
3. **Monitor Logs**: Regularly check service logs to detect and resolve issues promptly
|
||||
4. **Test Failover**: Regularly test registry failover mechanisms
|
||||
|
||||
This design is particularly suitable for large-scale production environment Kubernetes cluster image management needs, greatly simplifying operational complexity through dynamic configuration features.
|
||||
This design is particularly suitable for large-scale production environment Kubernetes cluster image management needs, greatly simplifying operational complexity through dynamic configuration features.
|
||||
|
||||
@@ -7,6 +7,7 @@ toolchain go1.23.1
|
||||
require (
|
||||
github.com/docker/docker v25.0.6+incompatible
|
||||
github.com/google/go-containerregistry v0.15.2
|
||||
github.com/hashicorp/golang-lru v0.5.4
|
||||
github.com/labring/sealos v0.0.0
|
||||
github.com/labring/sreg v0.1.7-rc3.0.20250728082818-441302dcb159
|
||||
github.com/pelletier/go-toml v1.9.5
|
||||
@@ -30,6 +31,7 @@ require (
|
||||
github.com/docker/docker-credential-helpers v0.7.0 // indirect
|
||||
github.com/docker/go-units v0.5.0 // indirect
|
||||
github.com/emicklei/go-restful/v3 v3.11.0 // indirect
|
||||
github.com/evanphx/json-patch v5.7.0+incompatible // indirect
|
||||
github.com/go-logr/logr v1.4.1 // indirect
|
||||
github.com/go-openapi/jsonpointer v0.19.6 // indirect
|
||||
github.com/go-openapi/jsonreference v0.20.2 // indirect
|
||||
|
||||
@@ -57,6 +57,8 @@ github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4
|
||||
github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
|
||||
github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g=
|
||||
github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
|
||||
github.com/evanphx/json-patch v5.7.0+incompatible h1:vgGkfT/9f8zE6tvSCe74nfpAVDQ2tG6yudJd8LBksgI=
|
||||
github.com/evanphx/json-patch v5.7.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
|
||||
github.com/felixge/httpsnoop v1.0.3 h1:s/nj+GCswXYzN5v2DpNMuMQYe+0DDwt5WVCU6CWBdXk=
|
||||
github.com/felixge/httpsnoop v1.0.3/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
|
||||
github.com/go-logr/logr v1.4.1 h1:pKouT5E8xu9zeFC39JXRDukb6JFQPXM5p5I91188VAQ=
|
||||
@@ -115,6 +117,8 @@ github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY
|
||||
github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
|
||||
github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
|
||||
github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
|
||||
github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
|
||||
github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
|
||||
github.com/huandu/xstrings v1.4.0 h1:D17IlohoQq4UcpqD7fDk80P7l+lwAmlFaBHgOipl2FU=
|
||||
github.com/huandu/xstrings v1.4.0/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE=
|
||||
github.com/imdario/mergo v0.3.16 h1:wwQJbIsHYGMUyLSPrEq1CT16AhnhNJQ51+4fdHUnCl4=
|
||||
|
||||
@@ -30,6 +30,7 @@ type AuthStore struct {
|
||||
mu sync.RWMutex
|
||||
criConfigs map[string]rtype.AuthConfig
|
||||
offlineCRIConfigs map[string]rtype.AuthConfig
|
||||
observers []func()
|
||||
}
|
||||
|
||||
func NewAuthStore(auth *types.ShimAuthConfig) *AuthStore {
|
||||
@@ -40,18 +41,25 @@ func NewAuthStore(auth *types.ShimAuthConfig) *AuthStore {
|
||||
|
||||
func (a *AuthStore) Update(auth *types.ShimAuthConfig) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
|
||||
if auth == nil {
|
||||
a.criConfigs = map[string]rtype.AuthConfig{}
|
||||
a.offlineCRIConfigs = map[string]rtype.AuthConfig{}
|
||||
logger.Warn("received empty shim auth config, cleared cached registry credentials")
|
||||
return
|
||||
}
|
||||
} else {
|
||||
a.criConfigs = cloneAuthMap(auth.CRIConfigs)
|
||||
a.offlineCRIConfigs = cloneAuthMap(auth.OfflineCRIConfigs)
|
||||
logger.Debug("updated shim auth config, registries: %d, offline: %d", len(a.criConfigs), len(a.offlineCRIConfigs))
|
||||
}
|
||||
|
||||
a.criConfigs = cloneAuthMap(auth.CRIConfigs)
|
||||
a.offlineCRIConfigs = cloneAuthMap(auth.OfflineCRIConfigs)
|
||||
logger.Info("updated shim auth config, registries: %d, offline: %d", len(a.criConfigs), len(a.offlineCRIConfigs))
|
||||
observers := append([]func(){}, a.observers...)
|
||||
a.mu.Unlock()
|
||||
|
||||
for _, observer := range observers {
|
||||
if observer != nil {
|
||||
observer()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (a *AuthStore) GetCRIConfig(registry string) (rtype.AuthConfig, bool) {
|
||||
@@ -76,6 +84,15 @@ func (a *AuthStore) GetOfflineConfigs() map[string]rtype.AuthConfig {
|
||||
return cloneAuthMap(a.offlineCRIConfigs)
|
||||
}
|
||||
|
||||
func (a *AuthStore) AddObserver(observer func()) {
|
||||
if observer == nil {
|
||||
return
|
||||
}
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
a.observers = append(a.observers, observer)
|
||||
}
|
||||
|
||||
func cloneAuthMap(src map[string]rtype.AuthConfig) map[string]rtype.AuthConfig {
|
||||
if len(src) == 0 {
|
||||
return map[string]rtype.AuthConfig{}
|
||||
|
||||
@@ -75,3 +75,23 @@ func TestAuthStoreUpdateAndGet(t *testing.T) {
|
||||
t.Fatalf("expected offline credentials to be cleared after nil update")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthStoreObservers(t *testing.T) {
|
||||
store := NewAuthStore(nil)
|
||||
calls := 0
|
||||
store.AddObserver(func() { calls++ })
|
||||
|
||||
store.Update(&types.ShimAuthConfig{
|
||||
CRIConfigs: map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {ServerAddress: "https://registry.example.com"},
|
||||
},
|
||||
})
|
||||
if calls != 1 {
|
||||
t.Fatalf("expected observer to fire once, got %d", calls)
|
||||
}
|
||||
|
||||
store.Update(nil)
|
||||
if calls != 2 {
|
||||
t.Fatalf("expected observer to fire on nil update, got %d", calls)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
// Copyright © 2025 sealos.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package server
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
rtype "github.com/docker/docker/api/types/registry"
|
||||
|
||||
"github.com/labring/image-cri-shim/pkg/types"
|
||||
)
|
||||
|
||||
func BenchmarkRewriteImageNoCache(b *testing.B) {
|
||||
withManifestStub(b, func(_ *manifestStub) {
|
||||
service := newImageServiceForTest(&types.ShimAuthConfig{
|
||||
CRIConfigs: map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {ServerAddress: "https://registry.example.com"},
|
||||
},
|
||||
})
|
||||
service.setMaxCacheSize(0)
|
||||
image := "registry.example.com/app/nginx:latest"
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
service.rewriteImage(image, "pull")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func BenchmarkRewriteImageCached(b *testing.B) {
|
||||
withManifestStub(b, func(_ *manifestStub) {
|
||||
service := newImageServiceForTest(&types.ShimAuthConfig{
|
||||
CRIConfigs: map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {ServerAddress: "https://registry.example.com"},
|
||||
},
|
||||
})
|
||||
service.setMaxCacheSize(defaultImageCacheSize)
|
||||
image := "registry.example.com/app/nginx:latest"
|
||||
// warm cache
|
||||
service.rewriteImage(image, "pull")
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
service.rewriteImage(image, "pull")
|
||||
}
|
||||
})
|
||||
}
|
||||
+459
-30
@@ -18,10 +18,14 @@ package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
rtype "github.com/docker/docker/api/types/registry"
|
||||
|
||||
"github.com/google/go-containerregistry/pkg/name"
|
||||
lru "github.com/hashicorp/golang-lru"
|
||||
|
||||
api "k8s.io/cri-api/pkg/apis/runtime/v1"
|
||||
|
||||
@@ -29,20 +33,419 @@ import (
|
||||
)
|
||||
|
||||
type v1ImageService struct {
|
||||
imageClient api.ImageServiceClient
|
||||
authStore *AuthStore
|
||||
imageClient api.ImageServiceClient
|
||||
authStore *AuthStore
|
||||
cacheMutex sync.RWMutex
|
||||
imageCache *lru.Cache
|
||||
domainCache *lru.Cache
|
||||
maxCacheSize int
|
||||
cacheTTL time.Duration
|
||||
domainTTL time.Duration
|
||||
metrics *cacheMetrics
|
||||
}
|
||||
|
||||
type cacheEntry struct {
|
||||
newImage string
|
||||
auth *rtype.AuthConfig
|
||||
found bool
|
||||
expiry time.Time
|
||||
}
|
||||
|
||||
type domainEntry struct {
|
||||
registryDomain string
|
||||
expiry time.Time
|
||||
}
|
||||
|
||||
type CacheOptions struct {
|
||||
ImageCacheSize int
|
||||
ImageCacheTTL time.Duration
|
||||
DomainCacheTTL time.Duration
|
||||
}
|
||||
|
||||
type CacheStats struct {
|
||||
ImageHits uint64
|
||||
ImageMisses uint64
|
||||
DomainHits uint64
|
||||
DomainMisses uint64
|
||||
ImageEvictions uint64
|
||||
DomainEvictions uint64
|
||||
Invalidations uint64
|
||||
GeneratedAt time.Time
|
||||
}
|
||||
|
||||
type cacheMetrics struct {
|
||||
imageHits atomic.Uint64
|
||||
imageMisses atomic.Uint64
|
||||
domainHits atomic.Uint64
|
||||
domainMisses atomic.Uint64
|
||||
imageEvictions atomic.Uint64
|
||||
domainEvictions atomic.Uint64
|
||||
invalidations atomic.Uint64
|
||||
}
|
||||
|
||||
const (
|
||||
// defaultImageCacheSize sets the default maximum number of cached image rewrite results.
|
||||
// 1024 entries is a balance between hit rate and memory use for typical clusters; at roughly
|
||||
// ~200 bytes per entry this uses around 200KB. Operators can tune this via config if they
|
||||
// need a larger or smaller cache.
|
||||
defaultImageCacheSize = 1024
|
||||
// defaultCacheTTL is how long cached rewrite results stay valid before being refreshed.
|
||||
// The 30m default keeps cache freshness reasonable without frequent churn; adjust if workloads
|
||||
// demand shorter or longer staleness windows.
|
||||
defaultCacheTTL = 30 * time.Minute
|
||||
// defaultDomainCacheTTL is the expiry window for registry-domain match entries.
|
||||
defaultDomainCacheTTL = 10 * time.Minute
|
||||
// domainCacheRatio determines the size of the domain cache as imageCacheSize/ratio. A ratio of
|
||||
// 10 keeps domain cache small relative to image cache to cap memory use while still providing
|
||||
// effective domain-level memoization; tweak if a different ratio is desired.
|
||||
domainCacheRatio = 10
|
||||
)
|
||||
|
||||
func (c CacheOptions) normalize() CacheOptions {
|
||||
if c.ImageCacheSize == 0 {
|
||||
c.ImageCacheSize = defaultImageCacheSize
|
||||
}
|
||||
if c.ImageCacheSize < 0 {
|
||||
c.ImageCacheSize = 0
|
||||
}
|
||||
if c.ImageCacheTTL <= 0 {
|
||||
c.ImageCacheTTL = defaultCacheTTL
|
||||
}
|
||||
if c.DomainCacheTTL <= 0 {
|
||||
c.DomainCacheTTL = defaultDomainCacheTTL
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
func newCacheMetrics() *cacheMetrics {
|
||||
return &cacheMetrics{}
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) snapshot() CacheStats {
|
||||
if m == nil {
|
||||
return CacheStats{}
|
||||
}
|
||||
return CacheStats{
|
||||
ImageHits: m.imageHits.Load(),
|
||||
ImageMisses: m.imageMisses.Load(),
|
||||
DomainHits: m.domainHits.Load(),
|
||||
DomainMisses: m.domainMisses.Load(),
|
||||
ImageEvictions: m.imageEvictions.Load(),
|
||||
DomainEvictions: m.domainEvictions.Load(),
|
||||
Invalidations: m.invalidations.Load(),
|
||||
GeneratedAt: time.Now(),
|
||||
}
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) recordImageHit() {
|
||||
if m != nil {
|
||||
m.imageHits.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) recordImageMiss() {
|
||||
if m != nil {
|
||||
m.imageMisses.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) recordDomainHit() {
|
||||
if m != nil {
|
||||
m.domainHits.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) recordDomainMiss() {
|
||||
if m != nil {
|
||||
m.domainMisses.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) recordImageEviction() {
|
||||
if m != nil {
|
||||
m.imageEvictions.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) recordDomainEviction() {
|
||||
if m != nil {
|
||||
m.domainEvictions.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) recordInvalidation() {
|
||||
if m != nil {
|
||||
m.invalidations.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
func newV1ImageService(client api.ImageServiceClient, authStore *AuthStore, cacheOpts CacheOptions) *v1ImageService {
|
||||
service := &v1ImageService{
|
||||
imageClient: client,
|
||||
authStore: authStore,
|
||||
metrics: newCacheMetrics(),
|
||||
}
|
||||
service.UpdateCacheOptions(cacheOpts)
|
||||
if authStore != nil {
|
||||
authStore.AddObserver(service.invalidateCache)
|
||||
}
|
||||
return service
|
||||
}
|
||||
|
||||
func (s *v1ImageService) UpdateCacheOptions(opts CacheOptions) {
|
||||
normalized := opts.normalize()
|
||||
s.cacheMutex.Lock()
|
||||
s.cacheTTL = normalized.ImageCacheTTL
|
||||
s.domainTTL = normalized.DomainCacheTTL
|
||||
s.cacheMutex.Unlock()
|
||||
s.setMaxCacheSize(normalized.ImageCacheSize)
|
||||
}
|
||||
|
||||
func (s *v1ImageService) rewriteImage(image, action string) (string, bool, *rtype.AuthConfig) {
|
||||
newImage, ok, auth := replaceImage(image, action, s.authStore.GetOfflineConfigs())
|
||||
if ok {
|
||||
return newImage, true, auth
|
||||
if entry, ok := s.getCachedResult(image); ok {
|
||||
s.logRewriteResult(action, image, entry.newImage, "cache", true, entry.found)
|
||||
return entry.newImage, entry.found, entry.auth
|
||||
}
|
||||
registries := s.authStore.GetCRIConfigs()
|
||||
if len(registries) == 0 {
|
||||
|
||||
if s.authStore == nil {
|
||||
s.logRewriteResult(action, image, image, "authstore-disabled", false, false)
|
||||
return image, false, nil
|
||||
}
|
||||
return replaceImage(image, action, registries)
|
||||
|
||||
// Prefer original domain first. If a matching registry is configured, try it;
|
||||
// only if that fails should we consider offline fallback elsewhere (handled by PullImage).
|
||||
domain := extractDomainFromImage(image)
|
||||
if domain != "" {
|
||||
if registries := s.authStore.GetCRIConfigs(); len(registries) > 0 {
|
||||
if matchedDomain, cfg := s.findMatchingRegistry(domain, registries); cfg != nil {
|
||||
newImage, ok, auth := replaceImage(image, action, map[string]rtype.AuthConfig{matchedDomain: *cfg})
|
||||
s.cacheResult(image, newImage, ok, auth)
|
||||
s.logRewriteResult(action, image, newImage, fmt.Sprintf("cri-domain:%s", matchedDomain), false, ok)
|
||||
if ok {
|
||||
return newImage, true, auth
|
||||
}
|
||||
// No manifest in matched registry: do not early return; let caller decide fallback.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Do not rewrite to offline here; leave image unchanged to let the runtime
|
||||
// attempt the original domain first. Offline fallback is performed on pull failure.
|
||||
s.cacheResult(image, image, false, nil)
|
||||
s.logRewriteResult(action, image, image, "no-rewrite", false, false)
|
||||
return image, false, nil
|
||||
}
|
||||
|
||||
func (s *v1ImageService) getCachedResult(image string) (cacheEntry, bool) {
|
||||
if image == "" {
|
||||
return cacheEntry{}, false
|
||||
}
|
||||
s.cacheMutex.Lock()
|
||||
defer s.cacheMutex.Unlock()
|
||||
if s.maxCacheSize <= 0 || s.imageCache == nil {
|
||||
return cacheEntry{}, false
|
||||
}
|
||||
|
||||
if raw, ok := s.imageCache.Get(image); ok {
|
||||
entry, valid := raw.(cacheEntry)
|
||||
if !valid || time.Now().After(entry.expiry) {
|
||||
s.imageCache.Remove(image)
|
||||
s.metrics.recordImageEviction()
|
||||
s.metrics.recordImageMiss()
|
||||
return cacheEntry{}, false
|
||||
}
|
||||
s.metrics.recordImageHit()
|
||||
entry.auth = cloneAuthConfig(entry.auth)
|
||||
return entry, true
|
||||
}
|
||||
s.metrics.recordImageMiss()
|
||||
return cacheEntry{}, false
|
||||
}
|
||||
|
||||
func (s *v1ImageService) cacheResult(image, newImage string, found bool, auth *rtype.AuthConfig) {
|
||||
if s.maxCacheSize <= 0 || image == "" {
|
||||
return
|
||||
}
|
||||
s.cacheMutex.Lock()
|
||||
defer s.cacheMutex.Unlock()
|
||||
if s.imageCache == nil {
|
||||
return
|
||||
}
|
||||
|
||||
ttl := s.cacheTTL
|
||||
if ttl <= 0 {
|
||||
ttl = defaultCacheTTL
|
||||
}
|
||||
|
||||
s.imageCache.Add(image, cacheEntry{
|
||||
newImage: newImage,
|
||||
auth: cloneAuthConfig(auth),
|
||||
found: found,
|
||||
expiry: time.Now().Add(ttl),
|
||||
})
|
||||
}
|
||||
|
||||
func (s *v1ImageService) cacheDomainMatch(imageDomain, registryDomain string) {
|
||||
if imageDomain == "" || registryDomain == "" {
|
||||
return
|
||||
}
|
||||
s.cacheMutex.Lock()
|
||||
defer s.cacheMutex.Unlock()
|
||||
if s.domainCache == nil {
|
||||
return
|
||||
}
|
||||
|
||||
ttl := s.domainTTL
|
||||
if ttl <= 0 {
|
||||
ttl = defaultDomainCacheTTL
|
||||
}
|
||||
s.domainCache.Add(imageDomain, domainEntry{
|
||||
registryDomain: registryDomain,
|
||||
expiry: time.Now().Add(ttl),
|
||||
})
|
||||
}
|
||||
|
||||
func (s *v1ImageService) getCachedDomainMatch(domain string, registries map[string]rtype.AuthConfig) (string, *rtype.AuthConfig) {
|
||||
s.cacheMutex.Lock()
|
||||
defer s.cacheMutex.Unlock()
|
||||
if s.domainCache == nil {
|
||||
return "", nil
|
||||
}
|
||||
raw, ok := s.domainCache.Get(domain)
|
||||
if !ok {
|
||||
s.metrics.recordDomainMiss()
|
||||
return "", nil
|
||||
}
|
||||
entry, valid := raw.(domainEntry)
|
||||
if !valid || time.Now().After(entry.expiry) {
|
||||
s.domainCache.Remove(domain)
|
||||
s.metrics.recordDomainEviction()
|
||||
s.metrics.recordDomainMiss()
|
||||
return "", nil
|
||||
}
|
||||
cfg, exists := registries[entry.registryDomain]
|
||||
if !exists {
|
||||
s.domainCache.Remove(domain)
|
||||
s.metrics.recordDomainEviction()
|
||||
s.metrics.recordDomainMiss()
|
||||
return "", nil
|
||||
}
|
||||
s.metrics.recordDomainHit()
|
||||
cfgCopy := cfg
|
||||
return entry.registryDomain, &cfgCopy
|
||||
}
|
||||
|
||||
func (s *v1ImageService) maxDomainCacheSize() int {
|
||||
if s.maxCacheSize <= 0 {
|
||||
return 0
|
||||
}
|
||||
size := s.maxCacheSize / domainCacheRatio
|
||||
if size == 0 {
|
||||
size = 1
|
||||
}
|
||||
return size
|
||||
}
|
||||
|
||||
func (s *v1ImageService) invalidateCache() {
|
||||
var purged bool
|
||||
s.cacheMutex.Lock()
|
||||
defer s.cacheMutex.Unlock()
|
||||
if s.imageCache != nil {
|
||||
s.imageCache.Purge()
|
||||
purged = true
|
||||
}
|
||||
if s.domainCache != nil {
|
||||
s.domainCache.Purge()
|
||||
purged = true
|
||||
}
|
||||
if purged {
|
||||
s.metrics.recordInvalidation()
|
||||
}
|
||||
}
|
||||
|
||||
func (s *v1ImageService) findMatchingRegistry(domain string, registries map[string]rtype.AuthConfig) (string, *rtype.AuthConfig) {
|
||||
if domain == "" || len(registries) == 0 {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
if cachedDomain, cfg := s.getCachedDomainMatch(domain, registries); cfg != nil {
|
||||
return cachedDomain, cfg
|
||||
}
|
||||
|
||||
for regDomain, cfg := range registries {
|
||||
if strings.EqualFold(domain, regDomain) {
|
||||
s.cacheDomainMatch(domain, regDomain)
|
||||
cfgCopy := cfg
|
||||
return regDomain, &cfgCopy
|
||||
}
|
||||
}
|
||||
|
||||
return "", nil
|
||||
}
|
||||
|
||||
func (s *v1ImageService) logRewriteResult(action, original, rewritten, source string, cacheHit bool, replaced bool) {
|
||||
logger.Info("rewrite action=%s cache_hit=%t source=%s original=%s result=%s replaced=%t",
|
||||
action, cacheHit, source, original, rewritten, replaced)
|
||||
}
|
||||
|
||||
func (s *v1ImageService) CacheStats() CacheStats {
|
||||
return s.metrics.snapshot()
|
||||
}
|
||||
|
||||
func (s *v1ImageService) setMaxCacheSize(size int) {
|
||||
s.cacheMutex.Lock()
|
||||
defer s.cacheMutex.Unlock()
|
||||
if s.maxCacheSize == size {
|
||||
if size == 0 || s.imageCache != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
s.maxCacheSize = size
|
||||
s.rebuildCachesLocked()
|
||||
}
|
||||
|
||||
func (s *v1ImageService) rebuildCachesLocked() {
|
||||
hadCache := s.imageCache != nil || s.domainCache != nil
|
||||
if s.maxCacheSize <= 0 {
|
||||
s.imageCache = nil
|
||||
s.domainCache = nil
|
||||
if hadCache {
|
||||
s.metrics.recordInvalidation()
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
imageCache, err := lru.New(s.maxCacheSize)
|
||||
if err != nil {
|
||||
logger.Warn("failed to create image cache: %v", err)
|
||||
s.imageCache = nil
|
||||
} else {
|
||||
s.imageCache = imageCache
|
||||
}
|
||||
|
||||
domainSize := s.maxDomainCacheSize()
|
||||
if domainSize <= 0 {
|
||||
s.domainCache = nil
|
||||
return
|
||||
}
|
||||
domainCache, err := lru.New(domainSize)
|
||||
if err != nil {
|
||||
logger.Warn("failed to create domain cache: %v", err)
|
||||
s.domainCache = nil
|
||||
return
|
||||
}
|
||||
s.domainCache = domainCache
|
||||
if hadCache {
|
||||
s.metrics.recordInvalidation()
|
||||
}
|
||||
}
|
||||
|
||||
func cloneAuthConfig(cfg *rtype.AuthConfig) *rtype.AuthConfig {
|
||||
if cfg == nil {
|
||||
return nil
|
||||
}
|
||||
copyCfg := *cfg
|
||||
return ©Cfg
|
||||
}
|
||||
|
||||
func ToV1AuthConfig(c *rtype.AuthConfig) *api.AuthConfig {
|
||||
@@ -90,32 +493,58 @@ func (s *v1ImageService) ImageStatus(ctx context.Context,
|
||||
}
|
||||
|
||||
func (s *v1ImageService) PullImage(ctx context.Context,
|
||||
req *api.PullImageRequest) (*api.PullImageResponse, error) {
|
||||
logger.Debug("PullImage begin: %+v", req)
|
||||
req *api.PullImageRequest) (*api.PullImageResponse, error) {
|
||||
logger.Debug("PullImage begin: %+v", req)
|
||||
if req.Image != nil {
|
||||
imageName := req.Image.Image
|
||||
if newImage, ok, auth := s.rewriteImage(req.Image.Image, "PullImage"); ok {
|
||||
imageName = newImage
|
||||
if auth != nil {
|
||||
req.Auth = ToV1AuthConfig(auth)
|
||||
}
|
||||
}
|
||||
if req.Auth == nil {
|
||||
ref, _ := name.ParseReference(imageName)
|
||||
registry := ref.Context().RegistryStr()
|
||||
if cfg, ok := s.authStore.GetCRIConfig(registry); ok {
|
||||
req.Auth = ToV1AuthConfig(&cfg)
|
||||
originalImage := req.Image.Image
|
||||
imageName := originalImage
|
||||
if req.Auth == nil && s.authStore != nil {
|
||||
if registries := s.authStore.GetCRIConfigs(); len(registries) > 0 {
|
||||
domain := extractDomainFromImage(imageName)
|
||||
if matchedDomain, cfg := s.findMatchingRegistry(domain, registries); cfg != nil {
|
||||
s.cacheDomainMatch(domain, matchedDomain)
|
||||
req.Auth = ToV1AuthConfig(cfg)
|
||||
}
|
||||
}
|
||||
}
|
||||
req.Image.Image = imageName
|
||||
}
|
||||
logger.Debug("PullImage after: %+v", req)
|
||||
rsp, err := s.imageClient.PullImage(ctx, req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
logger.Debug("PullImage after: %+v", req)
|
||||
rsp, err := s.imageClient.PullImage(ctx, req)
|
||||
if err == nil {
|
||||
return rsp, nil
|
||||
}
|
||||
logger.Warn("PullImage first attempt failed: %v", err)
|
||||
|
||||
return rsp, err
|
||||
// On failure, attempt offline/fallback rewrite and retry once.
|
||||
if req.Image != nil && s.authStore != nil {
|
||||
original := req.Image.Image
|
||||
// Do not rewrite for private registries configured in registries; honor original domain
|
||||
if registries := s.authStore.GetCRIConfigs(); len(registries) > 0 {
|
||||
if domain := extractDomainFromImage(original); domain != "" {
|
||||
if matchedDomain, cfg := s.findMatchingRegistry(domain, registries); cfg != nil {
|
||||
s.cacheDomainMatch(domain, matchedDomain)
|
||||
s.logRewriteResult("PullImageFallback", original, original, "private-domain-skip", false, false)
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
if newImage, ok, auth := replaceImage(original, "PullImageFallback", s.authStore.GetOfflineConfigs()); ok {
|
||||
s.cacheResult(original, newImage, ok, auth)
|
||||
s.logRewriteResult("PullImageFallback", original, newImage, "offline-manifest", false, true)
|
||||
req.Image.Image = newImage
|
||||
if auth != nil {
|
||||
req.Auth = ToV1AuthConfig(auth)
|
||||
}
|
||||
rsp2, err2 := s.imageClient.PullImage(ctx, req)
|
||||
if err2 == nil {
|
||||
return rsp2, nil
|
||||
}
|
||||
logger.Warn("PullImage fallback attempt failed: %v", err2)
|
||||
return nil, err2
|
||||
}
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
func (s *v1ImageService) RemoveImage(ctx context.Context,
|
||||
|
||||
+445
@@ -0,0 +1,445 @@
|
||||
// Copyright © 2025 sealos.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
rtype "github.com/docker/docker/api/types/registry"
|
||||
|
||||
"github.com/labring/image-cri-shim/pkg/types"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
api "k8s.io/cri-api/pkg/apis/runtime/v1"
|
||||
)
|
||||
|
||||
func TestRewriteImageDomainMatching(t *testing.T) {
|
||||
withManifestStub(t, func(_ *manifestStub) {
|
||||
tests := []struct {
|
||||
name string
|
||||
image string
|
||||
criConfigs map[string]rtype.AuthConfig
|
||||
offlineConfigs map[string]rtype.AuthConfig
|
||||
expectedDomain string
|
||||
expectedFound bool
|
||||
}{
|
||||
{
|
||||
name: "exact domain match",
|
||||
image: "registry.example.com/app/nginx:1.0",
|
||||
criConfigs: map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {ServerAddress: "https://registry.example.com"},
|
||||
},
|
||||
expectedDomain: "registry.example.com",
|
||||
expectedFound: true,
|
||||
},
|
||||
{
|
||||
name: "subdomain rewrite uses configured registry",
|
||||
image: "cache.registry.example.com/app/nginx:2.0",
|
||||
criConfigs: map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {ServerAddress: "https://registry.example.com"},
|
||||
},
|
||||
expectedDomain: "registry.example.com",
|
||||
expectedFound: true,
|
||||
},
|
||||
{
|
||||
name: "default registry falls back to offline",
|
||||
image: "nginx:latest",
|
||||
offlineConfigs: map[string]rtype.AuthConfig{
|
||||
"offline.registry.local": {ServerAddress: "https://offline.registry.local"},
|
||||
},
|
||||
expectedDomain: "offline.registry.local",
|
||||
expectedFound: true,
|
||||
},
|
||||
{
|
||||
name: "no registries available",
|
||||
image: "busybox:latest",
|
||||
criConfigs: map[string]rtype.AuthConfig{},
|
||||
offlineConfigs: map[string]rtype.AuthConfig{},
|
||||
expectedDomain: "",
|
||||
expectedFound: false,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
service := newImageServiceForTest(&types.ShimAuthConfig{
|
||||
CRIConfigs: tt.criConfigs,
|
||||
OfflineCRIConfigs: tt.offlineConfigs,
|
||||
})
|
||||
result, found, _ := service.rewriteImage(tt.image, "pull")
|
||||
if found != tt.expectedFound {
|
||||
t.Fatalf("expected found=%v got=%v", tt.expectedFound, found)
|
||||
}
|
||||
if tt.expectedDomain == "" {
|
||||
if result != tt.image {
|
||||
t.Fatalf("expected image unchanged, got %s", result)
|
||||
}
|
||||
return
|
||||
}
|
||||
if domain := extractDomainFromImage(result); domain != tt.expectedDomain {
|
||||
t.Fatalf("expected domain %q, got %q", tt.expectedDomain, domain)
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestFindMatchingRegistryExactMatchOnly(t *testing.T) {
|
||||
service := newV1ImageService(nil, nil, CacheOptions{})
|
||||
registries := map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {ServerAddress: "https://registry.example.com"},
|
||||
}
|
||||
|
||||
if domain, cfg := service.findMatchingRegistry("registry.example.com", registries); cfg == nil || domain != "registry.example.com" {
|
||||
t.Fatalf("expected exact match for registry.example.com, got domain=%q cfg=%v", domain, cfg)
|
||||
}
|
||||
if domain, cfg := service.findMatchingRegistry("REGISTRY.EXAMPLE.COM", registries); cfg == nil || domain != "registry.example.com" {
|
||||
t.Fatalf("expected case-insensitive exact match, got domain=%q cfg=%v", domain, cfg)
|
||||
}
|
||||
if domain, cfg := service.findMatchingRegistry("mirror.registry.example.com", registries); cfg != nil || domain != "" {
|
||||
t.Fatalf("expected no match for subdomain lookup, got domain=%q cfg=%v", domain, cfg)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRewriteImageCaching(t *testing.T) {
|
||||
withManifestStub(t, func(stub *manifestStub) {
|
||||
service := newImageServiceForTest(&types.ShimAuthConfig{
|
||||
CRIConfigs: map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {ServerAddress: "https://registry.example.com"},
|
||||
},
|
||||
})
|
||||
|
||||
image := "registry.example.com/app/nginx:1.0"
|
||||
if _, found, _ := service.rewriteImage(image, "pull"); !found {
|
||||
t.Fatalf("expected first rewrite to succeed")
|
||||
}
|
||||
if count := stub.callCount(image); count != 1 {
|
||||
t.Fatalf("expected 1 manifest call, got %d", count)
|
||||
}
|
||||
|
||||
if _, found, _ := service.rewriteImage(image, "pull"); !found {
|
||||
t.Fatalf("expected cached rewrite to succeed")
|
||||
}
|
||||
if count := stub.callCount(image); count != 1 {
|
||||
t.Fatalf("expected cache hit to avoid manifest call, got %d", count)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestRewriteImageCacheExpiry(t *testing.T) {
|
||||
withManifestStub(t, func(stub *manifestStub) {
|
||||
service := newImageServiceForTest(&types.ShimAuthConfig{
|
||||
CRIConfigs: map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {ServerAddress: "https://registry.example.com"},
|
||||
},
|
||||
})
|
||||
service.cacheTTL = 30 * time.Millisecond
|
||||
|
||||
image := "registry.example.com/app/nginx:latest"
|
||||
if _, found, _ := service.rewriteImage(image, "pull"); !found {
|
||||
t.Fatalf("expected rewrite to succeed")
|
||||
}
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
if _, found, _ := service.rewriteImage(image, "pull"); !found {
|
||||
t.Fatalf("expected rewrite after expiry to succeed")
|
||||
}
|
||||
|
||||
if count := stub.callCount(image); count != 2 {
|
||||
t.Fatalf("expected cache expiry to trigger new manifest call, got %d", count)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestRewriteImageConcurrentAccess(t *testing.T) {
|
||||
withManifestStub(t, func(_ *manifestStub) {
|
||||
service := newImageServiceForTest(&types.ShimAuthConfig{
|
||||
CRIConfigs: map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {ServerAddress: "https://registry.example.com"},
|
||||
},
|
||||
})
|
||||
service.cacheTTL = time.Second
|
||||
|
||||
const goroutines = 20
|
||||
const iterations = 20
|
||||
image := "registry.example.com/app/nginx:1.1"
|
||||
|
||||
var wg sync.WaitGroup
|
||||
results := make(chan string, goroutines*iterations)
|
||||
errCh := make(chan error, goroutines)
|
||||
for i := 0; i < goroutines; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for j := 0; j < iterations; j++ {
|
||||
out, found, _ := service.rewriteImage(image, "pull")
|
||||
if !found {
|
||||
errCh <- fmt.Errorf("expected rewrite to succeed in concurrent workload")
|
||||
return
|
||||
}
|
||||
results <- out
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
close(results)
|
||||
close(errCh)
|
||||
|
||||
if err := <-errCh; err != nil {
|
||||
t.Fatalf("concurrent rewrite error: %v", err)
|
||||
}
|
||||
|
||||
var expected string
|
||||
for result := range results {
|
||||
if expected == "" {
|
||||
expected = result
|
||||
continue
|
||||
}
|
||||
if result != expected {
|
||||
t.Fatalf("expected %s, got %s", expected, result)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestImageCacheEvictionOrder(t *testing.T) {
|
||||
withManifestStub(t, func(stub *manifestStub) {
|
||||
service := newImageServiceForTest(&types.ShimAuthConfig{
|
||||
CRIConfigs: map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {ServerAddress: "https://registry.example.com"},
|
||||
},
|
||||
})
|
||||
service.setMaxCacheSize(2)
|
||||
|
||||
images := []string{
|
||||
"registry.example.com/app/a:1",
|
||||
"registry.example.com/app/b:1",
|
||||
"registry.example.com/app/c:1",
|
||||
}
|
||||
|
||||
for i := 0; i < 2; i++ {
|
||||
if _, found, _ := service.rewriteImage(images[i], "pull"); !found {
|
||||
t.Fatalf("expected rewrite for %s to succeed", images[i])
|
||||
}
|
||||
if got := stub.callCount(images[i]); got != 1 {
|
||||
t.Fatalf("expected manifest call for %s to be recorded once, got %d", images[i], got)
|
||||
}
|
||||
}
|
||||
|
||||
// Touch the first image so the second one becomes the LRU entry.
|
||||
if _, found, _ := service.rewriteImage(images[0], "pull"); !found {
|
||||
t.Fatalf("expected cache hit for %s", images[0])
|
||||
}
|
||||
if got := stub.callCount(images[0]); got != 1 {
|
||||
t.Fatalf("expected cache hit to avoid manifest call for %s, got %d", images[0], got)
|
||||
}
|
||||
|
||||
if _, found, _ := service.rewriteImage(images[2], "pull"); !found {
|
||||
t.Fatalf("expected rewrite for %s to succeed", images[2])
|
||||
}
|
||||
|
||||
if _, found, _ := service.rewriteImage(images[1], "pull"); !found {
|
||||
t.Fatalf("expected rewrite for %s to succeed", images[1])
|
||||
}
|
||||
if got := stub.callCount(images[1]); got != 2 {
|
||||
t.Fatalf("expected LRU eviction to trigger second manifest call for %s, got %d", images[1], got)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestDomainCacheExpiry(t *testing.T) {
|
||||
service := newV1ImageService(nil, nil, CacheOptions{})
|
||||
service.setMaxCacheSize(4)
|
||||
service.domainTTL = 20 * time.Millisecond
|
||||
|
||||
registries := map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {ServerAddress: "https://registry.example.com"},
|
||||
}
|
||||
|
||||
service.cacheDomainMatch("mirror.registry.example.com", "registry.example.com")
|
||||
if domain, cfg := service.getCachedDomainMatch("mirror.registry.example.com", registries); cfg == nil || domain != "registry.example.com" {
|
||||
t.Fatalf("expected domain cache hit, got domain=%s cfg=%v", domain, cfg)
|
||||
}
|
||||
|
||||
time.Sleep(40 * time.Millisecond)
|
||||
|
||||
if domain, cfg := service.getCachedDomainMatch("mirror.registry.example.com", registries); cfg != nil || domain != "" {
|
||||
t.Fatalf("expected domain cache entry to expire, got domain=%s cfg=%v", domain, cfg)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCacheStatsAccounting(t *testing.T) {
|
||||
withManifestStub(t, func(_ *manifestStub) {
|
||||
service := newImageServiceForTest(&types.ShimAuthConfig{
|
||||
CRIConfigs: map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {ServerAddress: "https://registry.example.com"},
|
||||
},
|
||||
})
|
||||
service.cacheTTL = 10 * time.Millisecond
|
||||
|
||||
image := "registry.example.com/app/nginx:latest"
|
||||
// first rewrite should miss cache and populate entries
|
||||
if _, found, _ := service.rewriteImage(image, "pull"); !found {
|
||||
t.Fatalf("expected rewrite to succeed")
|
||||
}
|
||||
stats := service.CacheStats()
|
||||
if stats.ImageMisses == 0 {
|
||||
t.Fatalf("expected image miss to be recorded, got %+v", stats)
|
||||
}
|
||||
if stats.DomainMisses == 0 {
|
||||
t.Fatalf("expected domain miss to be recorded, got %+v", stats)
|
||||
}
|
||||
|
||||
// second rewrite should hit image cache
|
||||
if _, found, _ := service.rewriteImage(image, "pull"); !found {
|
||||
t.Fatalf("expected rewrite to succeed")
|
||||
}
|
||||
stats = service.CacheStats()
|
||||
if stats.ImageHits == 0 {
|
||||
t.Fatalf("expected cache hits to be recorded, got %+v", stats)
|
||||
}
|
||||
|
||||
// simulate domain cache hit by looking up cached mapping directly
|
||||
service.cacheDomainMatch("mirror.registry.example.com", "registry.example.com")
|
||||
if _, cfg := service.getCachedDomainMatch("mirror.registry.example.com", map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {ServerAddress: "https://registry.example.com"},
|
||||
}); cfg == nil {
|
||||
t.Fatalf("expected domain cache hit")
|
||||
}
|
||||
stats = service.CacheStats()
|
||||
if stats.DomainHits == 0 {
|
||||
t.Fatalf("expected domain hit counter to increment, got %+v", stats)
|
||||
}
|
||||
|
||||
// force expiry to trigger eviction counters
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
if _, found, _ := service.rewriteImage(image, "pull"); !found {
|
||||
t.Fatalf("expected rewrite to succeed after expiry")
|
||||
}
|
||||
stats = service.CacheStats()
|
||||
if stats.ImageEvictions == 0 {
|
||||
t.Fatalf("expected eviction counter to increment, got %+v", stats)
|
||||
}
|
||||
|
||||
// invalidation should bump invalidation counter
|
||||
service.invalidateCache()
|
||||
stats = service.CacheStats()
|
||||
if stats.Invalidations == 0 {
|
||||
t.Fatalf("expected invalidation counter to increment, got %+v", stats)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestRewriteImageEdgeCases(t *testing.T) {
|
||||
withManifestStub(t, func(_ *manifestStub) {
|
||||
service := newImageServiceForTest(nil)
|
||||
|
||||
if image, found, _ := service.rewriteImage("", "pull"); found || image != "" {
|
||||
t.Fatalf("expected empty image to remain unchanged, got %q found=%v", image, found)
|
||||
}
|
||||
|
||||
if image, found, _ := service.rewriteImage("bad@@@", "pull"); found {
|
||||
t.Fatalf("expected invalid image to fail rewrite, got %q", image)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestPullImageInjectsAuthFromCRIConfig(t *testing.T) {
|
||||
withManifestStub(t, func(_ *manifestStub) {
|
||||
tests := []struct {
|
||||
name string
|
||||
registries map[string]rtype.AuthConfig
|
||||
image string
|
||||
wantUser string
|
||||
}{
|
||||
{
|
||||
name: "exact domain match",
|
||||
registries: map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {
|
||||
Username: "bot",
|
||||
Password: "token",
|
||||
ServerAddress: "https://registry.example.com",
|
||||
},
|
||||
},
|
||||
image: "registry.example.com/app/nginx:latest",
|
||||
wantUser: "bot",
|
||||
},
|
||||
{
|
||||
name: "docker hub alias",
|
||||
registries: map[string]rtype.AuthConfig{
|
||||
"registry-1.docker.io": {
|
||||
Username: "hubuser",
|
||||
Password: "token",
|
||||
ServerAddress: "https://registry-1.docker.io",
|
||||
},
|
||||
},
|
||||
image: "nginx:latest",
|
||||
wantUser: "hubuser",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
store := NewAuthStore(&types.ShimAuthConfig{CRIConfigs: tt.registries})
|
||||
client := &fakeImageClient{}
|
||||
service := newV1ImageService(client, store, CacheOptions{})
|
||||
req := &api.PullImageRequest{
|
||||
Image: &api.ImageSpec{Image: tt.image},
|
||||
}
|
||||
if _, err := service.PullImage(context.Background(), req); err != nil {
|
||||
t.Fatalf("PullImage failed: %v", err)
|
||||
}
|
||||
if client.lastPull == nil {
|
||||
t.Fatalf("expected client to observe pull request")
|
||||
}
|
||||
if client.lastPull.Auth == nil || client.lastPull.Auth.Username != tt.wantUser {
|
||||
t.Fatalf("expected auth user %q, got %+v", tt.wantUser, client.lastPull.Auth)
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
type fakeImageClient struct {
|
||||
lastPull *api.PullImageRequest
|
||||
}
|
||||
|
||||
func (f *fakeImageClient) ListImages(ctx context.Context, in *api.ListImagesRequest, opts ...grpc.CallOption) (*api.ListImagesResponse, error) {
|
||||
return &api.ListImagesResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *fakeImageClient) ImageStatus(ctx context.Context, in *api.ImageStatusRequest, opts ...grpc.CallOption) (*api.ImageStatusResponse, error) {
|
||||
return &api.ImageStatusResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *fakeImageClient) PullImage(ctx context.Context, in *api.PullImageRequest, opts ...grpc.CallOption) (*api.PullImageResponse, error) {
|
||||
f.lastPull = in
|
||||
ref := ""
|
||||
if in.GetImage() != nil {
|
||||
ref = in.GetImage().GetImage()
|
||||
}
|
||||
return &api.PullImageResponse{ImageRef: ref}, nil
|
||||
}
|
||||
|
||||
func (f *fakeImageClient) RemoveImage(ctx context.Context, in *api.RemoveImageRequest, opts ...grpc.CallOption) (*api.RemoveImageResponse, error) {
|
||||
return &api.RemoveImageResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *fakeImageClient) ImageFsInfo(ctx context.Context, in *api.ImageFsInfoRequest, opts ...grpc.CallOption) (*api.ImageFsInfoResponse, error) {
|
||||
return &api.ImageFsInfoResponse{}, nil
|
||||
}
|
||||
+63
@@ -0,0 +1,63 @@
|
||||
// Copyright © 2025 sealos.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package server
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
rtype "github.com/docker/docker/api/types/registry"
|
||||
|
||||
"github.com/labring/image-cri-shim/pkg/types"
|
||||
)
|
||||
|
||||
func TestConfigurationUpdateInvalidatesCache(t *testing.T) {
|
||||
withManifestStub(t, func(stub *manifestStub) {
|
||||
store := NewAuthStore(&types.ShimAuthConfig{
|
||||
CRIConfigs: map[string]rtype.AuthConfig{
|
||||
"registry.example.com": {ServerAddress: "https://registry.example.com"},
|
||||
},
|
||||
})
|
||||
service := newV1ImageService(nil, store, CacheOptions{})
|
||||
service.setMaxCacheSize(4)
|
||||
service.cacheTTL = 500 * time.Millisecond
|
||||
|
||||
image := "registry.example.com/app/nginx:1.0"
|
||||
first, found, _ := service.rewriteImage(image, "pull")
|
||||
if !found {
|
||||
t.Fatalf("expected initial rewrite to succeed")
|
||||
}
|
||||
if domain := extractDomainFromImage(first); domain != "registry.example.com" {
|
||||
t.Fatalf("expected initial domain registry.example.com, got %s", domain)
|
||||
}
|
||||
|
||||
store.Update(&types.ShimAuthConfig{
|
||||
CRIConfigs: map[string]rtype.AuthConfig{
|
||||
"mirror.example.com": {ServerAddress: "https://mirror.example.com"},
|
||||
},
|
||||
})
|
||||
|
||||
second, found, _ := service.rewriteImage(image, "pull")
|
||||
if !found {
|
||||
t.Fatalf("expected rewrite after update to succeed")
|
||||
}
|
||||
if domain := extractDomainFromImage(second); domain != "mirror.example.com" {
|
||||
t.Fatalf("expected rewritten image to use updated domain, got %s", domain)
|
||||
}
|
||||
if count := stub.callCount(image); count != 2 {
|
||||
t.Fatalf("expected cache invalidation to force new manifest call, got %d", count)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -32,17 +32,19 @@ import (
|
||||
)
|
||||
|
||||
type Options struct {
|
||||
Timeout time.Duration
|
||||
// Socket is the socket where shim listens on
|
||||
Socket string
|
||||
// User is the user ID for our gRPC socket.
|
||||
User int
|
||||
// Group is the group ID for our gRPC socket.
|
||||
Group int
|
||||
// Mode is the permission mode bits for our gRPC socket.
|
||||
Mode os.FileMode
|
||||
// AuthStore keeps registry credentials shared with the CRI handlers.
|
||||
AuthStore *AuthStore
|
||||
Timeout time.Duration
|
||||
// Socket is the socket where shim listens on
|
||||
Socket string
|
||||
// User is the user ID for our gRPC socket.
|
||||
User int
|
||||
// Group is the group ID for our gRPC socket.
|
||||
Group int
|
||||
// Mode is the permission mode bits for our gRPC socket.
|
||||
Mode os.FileMode
|
||||
// AuthStore keeps registry credentials shared with the CRI handlers.
|
||||
AuthStore *AuthStore
|
||||
// Cache keeps cache tuning knobs.
|
||||
Cache CacheOptions
|
||||
}
|
||||
|
||||
type Server interface {
|
||||
@@ -55,11 +57,16 @@ type Server interface {
|
||||
Start() error
|
||||
|
||||
Stop()
|
||||
|
||||
UpdateCacheOptions(CacheOptions)
|
||||
|
||||
CacheStats() CacheStats
|
||||
}
|
||||
|
||||
type server struct {
|
||||
server *grpc.Server
|
||||
imageV1Client k8sv1api.ImageServiceClient
|
||||
imageService *v1ImageService
|
||||
options Options
|
||||
listener net.Listener // socket our gRPC server listens on
|
||||
}
|
||||
@@ -76,10 +83,9 @@ func (s *server) RegisterImageService(conn *grpc.ClientConn) error {
|
||||
return err
|
||||
}
|
||||
|
||||
k8sv1api.RegisterImageServiceServer(s.server, &v1ImageService{
|
||||
imageClient: s.imageV1Client,
|
||||
authStore: s.options.AuthStore,
|
||||
})
|
||||
imageService := newV1ImageService(s.imageV1Client, s.options.AuthStore, s.options.Cache)
|
||||
k8sv1api.RegisterImageServiceServer(s.server, imageService)
|
||||
s.imageService = imageService
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -189,6 +195,21 @@ func (s *server) Stop() {
|
||||
s.server.Stop()
|
||||
}
|
||||
|
||||
func (s *server) UpdateCacheOptions(opts CacheOptions) {
|
||||
if s.imageService == nil {
|
||||
logger.Warn("image service not initialized, skip cache update")
|
||||
return
|
||||
}
|
||||
s.imageService.UpdateCacheOptions(opts)
|
||||
}
|
||||
|
||||
func (s *server) CacheStats() CacheStats {
|
||||
if s.imageService == nil {
|
||||
return CacheStats{}
|
||||
}
|
||||
return s.imageService.CacheStats()
|
||||
}
|
||||
|
||||
func NewServer(options Options) (Server, error) {
|
||||
if !filepath.IsAbs(options.Socket) {
|
||||
return nil, fmt.Errorf("invalid socked")
|
||||
|
||||
+92
@@ -0,0 +1,92 @@
|
||||
// Copyright © 2025 sealos.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package server
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
rtype "github.com/docker/docker/api/types/registry"
|
||||
|
||||
name "github.com/google/go-containerregistry/pkg/name"
|
||||
|
||||
"github.com/labring/image-cri-shim/pkg/types"
|
||||
)
|
||||
|
||||
type manifestStub struct {
|
||||
mu sync.Mutex
|
||||
calls map[string]int
|
||||
}
|
||||
|
||||
func (m *manifestStub) handler(image string, auth map[string]rtype.AuthConfig) (string, []byte, *rtype.AuthConfig, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if m.calls == nil {
|
||||
m.calls = make(map[string]int)
|
||||
}
|
||||
m.calls[image]++
|
||||
|
||||
if len(auth) == 0 {
|
||||
return image, nil, nil, fmt.Errorf("no auth config provided")
|
||||
}
|
||||
|
||||
ref, err := name.ParseReference(image)
|
||||
if err != nil {
|
||||
return "", nil, nil, err
|
||||
}
|
||||
domain, cfg := pickFirstAuth(auth)
|
||||
|
||||
newImage := domain + "/" + ref.Context().RepositoryStr() + referenceSuffix(ref)
|
||||
cfgCopy := cfg
|
||||
return newImage, []byte("stub"), &cfgCopy, nil
|
||||
}
|
||||
|
||||
func (m *manifestStub) callCount(image string) int {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.calls[image]
|
||||
}
|
||||
|
||||
func pickFirstAuth(auth map[string]rtype.AuthConfig) (string, rtype.AuthConfig) {
|
||||
keys := make([]string, 0, len(auth))
|
||||
for key := range auth {
|
||||
keys = append(keys, key)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
key := keys[0]
|
||||
return key, auth[key]
|
||||
}
|
||||
|
||||
func withManifestStub(tb testing.TB, fn func(*manifestStub)) {
|
||||
tb.Helper()
|
||||
original := craneGetImageManifest
|
||||
stub := &manifestStub{}
|
||||
craneGetImageManifest = stub.handler
|
||||
defer func() {
|
||||
craneGetImageManifest = original
|
||||
}()
|
||||
fn(stub)
|
||||
}
|
||||
|
||||
func newImageServiceForTest(auth *types.ShimAuthConfig) *v1ImageService {
|
||||
store := NewAuthStore(auth)
|
||||
service := newV1ImageService(nil, store, CacheOptions{})
|
||||
service.setMaxCacheSize(32)
|
||||
service.cacheTTL = 200 * time.Millisecond
|
||||
return service
|
||||
}
|
||||
@@ -17,17 +17,20 @@ limitations under the License.
|
||||
package server
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"strings"
|
||||
|
||||
"github.com/docker/docker/api/types/registry"
|
||||
"github.com/docker/docker/api/types/registry"
|
||||
|
||||
"github.com/labring/sreg/pkg/registry/crane"
|
||||
"github.com/labring/sreg/pkg/registry/crane"
|
||||
|
||||
"github.com/labring/sealos/pkg/utils/logger"
|
||||
"github.com/labring/sealos/pkg/utils/logger"
|
||||
|
||||
name "github.com/google/go-containerregistry/pkg/name"
|
||||
name "github.com/google/go-containerregistry/pkg/name"
|
||||
)
|
||||
|
||||
// craneGetImageManifest is declared as a var so tests can replace it with a stub implementation.
|
||||
var craneGetImageManifest = crane.GetImageManifestFromAuth
|
||||
|
||||
// replaceImage replaces the image name to a new valid image name with the private registry.
|
||||
func replaceImage(image, action string, authConfig map[string]registry.AuthConfig) (newImage string,
|
||||
isReplace bool, cfg *registry.AuthConfig) {
|
||||
@@ -37,7 +40,7 @@ func replaceImage(image, action string, authConfig map[string]registry.AuthConfi
|
||||
|
||||
ref, err := name.ParseReference(image)
|
||||
if err != nil {
|
||||
logger.Warn("failed to parse image reference %s: %v", image, err)
|
||||
logger.Warn("rewrite skipped: unable to parse image reference %s: %v", image, err)
|
||||
return image, false, nil
|
||||
}
|
||||
|
||||
@@ -47,16 +50,14 @@ func replaceImage(image, action string, authConfig map[string]registry.AuthConfi
|
||||
return image, false, nil
|
||||
}
|
||||
|
||||
newImage, _, cfg, err = crane.GetImageManifestFromAuth(image, authConfig)
|
||||
newImage, _, cfg, err = craneGetImageManifest(image, authConfig)
|
||||
if err != nil {
|
||||
if strings.Contains(image, "@") {
|
||||
return replaceImage(strings.Split(image, "@")[0], action, authConfig)
|
||||
}
|
||||
logger.Warn("get image %s manifest error %s", newImage, err.Error())
|
||||
logger.Debug("image %s not found in registry, skipping", image)
|
||||
logger.Warn("rewrite skipped: failed to fetch manifest for %s (action=%s): %v", image, action, err)
|
||||
return image, false, cfg
|
||||
}
|
||||
logger.Info("image: %s, newImage: %s, action: %s", image, newImage, action)
|
||||
return newImage, true, cfg
|
||||
}
|
||||
|
||||
@@ -91,3 +92,31 @@ func referenceSuffix(ref name.Reference) string {
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
const defaultDockerRegistry = "docker.io"
|
||||
|
||||
func extractDomainFromImage(image string) string {
|
||||
if image == "" {
|
||||
return ""
|
||||
}
|
||||
ref, err := name.ParseReference(image)
|
||||
if err != nil {
|
||||
return normalizeDomainCandidate(registryFromImage(image))
|
||||
}
|
||||
domain := ref.Context().RegistryStr()
|
||||
if domain == "" {
|
||||
return defaultDockerRegistry
|
||||
}
|
||||
return normalizeDomainCandidate(domain)
|
||||
}
|
||||
|
||||
func normalizeDomainCandidate(domain string) string {
|
||||
switch domain {
|
||||
case "", "library", "docker", "index.docker.io", "registry-1.docker.io":
|
||||
return defaultDockerRegistry
|
||||
}
|
||||
if !strings.Contains(domain, ".") && !strings.Contains(domain, ":") && domain != "localhost" {
|
||||
return defaultDockerRegistry
|
||||
}
|
||||
return domain
|
||||
}
|
||||
|
||||
@@ -23,26 +23,42 @@ import (
|
||||
)
|
||||
|
||||
func TestReplaceImageSkipLogin(t *testing.T) {
|
||||
image := "nginx:1.29.2-alpine3.22-perl"
|
||||
authConfig := map[string]registry.AuthConfig{
|
||||
"docker.xpg666.xyz": {
|
||||
ServerAddress: "https://docker.xpg666.xyz",
|
||||
},
|
||||
"192.168.64.4:5000": {
|
||||
ServerAddress: "http://192.168.64.4:5000",
|
||||
Username: "admin",
|
||||
Password: "passw0rd",
|
||||
},
|
||||
withManifestStub(t, func(_ *manifestStub) {
|
||||
image := "nginx:1.29.2-alpine3.22-perl"
|
||||
authConfig := map[string]registry.AuthConfig{
|
||||
"docker.xpg666.xyz": {
|
||||
ServerAddress: "https://docker.xpg666.xyz",
|
||||
},
|
||||
}
|
||||
|
||||
newImage, replaced, cfg := replaceImage(image, "PullImage", authConfig)
|
||||
if !replaced {
|
||||
t.Fatalf("expected image to be replaced when login is skipped")
|
||||
}
|
||||
if expected := "docker.xpg666.xyz/library/nginx:1.29.2-alpine3.22-perl"; newImage != expected {
|
||||
t.Fatalf("expected rewritten image %q, got %q", expected, newImage)
|
||||
}
|
||||
if cfg == nil {
|
||||
t.Fatal("expected auth config to be returned")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestExtractDomainFromImage(t *testing.T) {
|
||||
tests := []struct {
|
||||
image string
|
||||
expected string
|
||||
}{
|
||||
{"nginx:latest", "docker.io"},
|
||||
{"docker.io/library/nginx:latest", "docker.io"},
|
||||
{"registry.example.com/app/nginx:1.0", "registry.example.com"},
|
||||
{"localhost:5000/nginx:1.0", "localhost:5000"},
|
||||
{"", ""},
|
||||
}
|
||||
|
||||
newImage, replaced, cfg := replaceImage(image, "PullImage", authConfig)
|
||||
if !replaced {
|
||||
t.Fatalf("expected image to be replaced when login is skipped")
|
||||
}
|
||||
if expected := "docker.xpg666.xyz/library/nginx:1.29.2-alpine3.22-perl"; newImage != expected {
|
||||
t.Fatalf("expected rewritten image %q, got %q", expected, newImage)
|
||||
}
|
||||
if cfg == nil {
|
||||
t.Fatal("expected no auth config")
|
||||
for _, tt := range tests {
|
||||
if domain := extractDomainFromImage(tt.image); domain != tt.expected {
|
||||
t.Fatalf("image %q expected domain %q, got %q", tt.image, tt.expected, domain)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -34,6 +34,9 @@ const (
|
||||
DisableService = server.DontConnect
|
||||
)
|
||||
|
||||
type CacheStats = server.CacheStats
|
||||
type CacheOptions = server.CacheOptions
|
||||
|
||||
// Shim is the interface we expose for controlling our CRI shim.
|
||||
type Shim interface {
|
||||
// Setup prepares the shim to start processing CRI requests.
|
||||
@@ -44,6 +47,10 @@ type Shim interface {
|
||||
Stop()
|
||||
// UpdateAuth refreshes registry credentials without restarting the shim.
|
||||
UpdateAuth(*types.ShimAuthConfig)
|
||||
// UpdateCache applies cache-related tuning knobs.
|
||||
UpdateCache(CacheOptions)
|
||||
// CacheStats returns current cache counters.
|
||||
CacheStats() CacheStats
|
||||
}
|
||||
|
||||
// shim is the implementation of Shim.
|
||||
@@ -73,14 +80,15 @@ func NewShim(cfg *types.Config, auth *types.ShimAuthConfig) (Shim, error) {
|
||||
|
||||
r.authStore = server.NewAuthStore(auth)
|
||||
|
||||
srvopts := server.Options{
|
||||
Timeout: cfg.Timeout.Duration,
|
||||
Socket: cfg.ImageShimSocket,
|
||||
User: -1,
|
||||
Group: -1,
|
||||
Mode: 0660,
|
||||
AuthStore: r.authStore,
|
||||
}
|
||||
srvopts := server.Options{
|
||||
Timeout: cfg.Timeout.Duration,
|
||||
Socket: cfg.ImageShimSocket,
|
||||
User: -1,
|
||||
Group: -1,
|
||||
Mode: 0660,
|
||||
AuthStore: r.authStore,
|
||||
Cache: CacheOptionsFromConfig(cfg),
|
||||
}
|
||||
srv, err := server.NewServer(srvopts)
|
||||
if err != nil {
|
||||
return nil, shimError("failed to create shim server: %v", err)
|
||||
@@ -127,6 +135,20 @@ func (r *shim) UpdateAuth(auth *types.ShimAuthConfig) {
|
||||
r.authStore.Update(auth)
|
||||
}
|
||||
|
||||
func (r *shim) UpdateCache(opts CacheOptions) {
|
||||
if r.server == nil {
|
||||
return
|
||||
}
|
||||
r.server.UpdateCacheOptions(opts)
|
||||
}
|
||||
|
||||
func (r *shim) CacheStats() CacheStats {
|
||||
if r.server == nil {
|
||||
return CacheStats{}
|
||||
}
|
||||
return r.server.CacheStats()
|
||||
}
|
||||
|
||||
func (r *shim) dialNotify(socket string, uid int, gid int, mode os.FileMode, err error) {
|
||||
if err != nil {
|
||||
logger.Error("failed to determine permissions/ownership of client socket %q: %v",
|
||||
@@ -147,3 +169,14 @@ func (r *shim) dialNotify(socket string, uid int, gid int, mode os.FileMode, err
|
||||
var shimError = func(format string, args ...interface{}) error {
|
||||
return fmt.Errorf("cri/shim: "+format, args...)
|
||||
}
|
||||
|
||||
func CacheOptionsFromConfig(cfg *types.Config) CacheOptions {
|
||||
if cfg == nil {
|
||||
return CacheOptions{}
|
||||
}
|
||||
return CacheOptions{
|
||||
ImageCacheSize: cfg.Cache.ImageCacheSize,
|
||||
ImageCacheTTL: cfg.Cache.ImageCacheTTL.Duration,
|
||||
DomainCacheTTL: cfg.Cache.DomainCacheTTL.Duration,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -39,6 +39,10 @@ const (
|
||||
SealosShimSock = "/var/run/image-cri-shim.sock"
|
||||
DefaultImageCRIShimConfig = "/etc/image-cri-shim.yaml"
|
||||
DefaultReloadInterval = 15 * time.Second
|
||||
defaultCacheStatsInterval = time.Minute
|
||||
defaultImageCacheTTL = 30 * time.Minute
|
||||
defaultDomainCacheTTL = 10 * time.Minute
|
||||
defaultImageCacheSize = 1024
|
||||
)
|
||||
|
||||
type Registry struct {
|
||||
@@ -47,15 +51,24 @@ type Registry struct {
|
||||
}
|
||||
|
||||
type Config struct {
|
||||
ImageShimSocket string `json:"shim"`
|
||||
RuntimeSocket string `json:"cri"`
|
||||
Address string `json:"address"`
|
||||
Force bool `json:"force"`
|
||||
Debug bool `json:"debug"`
|
||||
Timeout metav1.Duration `json:"timeout"`
|
||||
ReloadInterval metav1.Duration `json:"reloadInterval"`
|
||||
Auth string `json:"auth"`
|
||||
Registries []Registry `json:"registries" yaml:"registries,omitempty"`
|
||||
ImageShimSocket string `json:"shim"`
|
||||
RuntimeSocket string `json:"cri"`
|
||||
Address string `json:"address"`
|
||||
Force bool `json:"force"`
|
||||
Debug bool `json:"debug"`
|
||||
Timeout metav1.Duration `json:"timeout"`
|
||||
ReloadInterval metav1.Duration `json:"reloadInterval"`
|
||||
Auth string `json:"auth"`
|
||||
Cache CacheConfig `json:"cache" yaml:"cache"`
|
||||
Registries []Registry `json:"registries" yaml:"registries,omitempty"`
|
||||
}
|
||||
|
||||
type CacheConfig struct {
|
||||
ImageCacheSize int `json:"imageCacheSize" yaml:"imageCacheSize"`
|
||||
ImageCacheTTL metav1.Duration `json:"imageCacheTTL" yaml:"imageCacheTTL"`
|
||||
DomainCacheTTL metav1.Duration `json:"domainCacheTTL" yaml:"domainCacheTTL"`
|
||||
StatsLogInterval metav1.Duration `json:"statsLogInterval" yaml:"statsLogInterval"`
|
||||
DisableStats bool `json:"disableStats" yaml:"disableStats"`
|
||||
}
|
||||
|
||||
type ShimAuthConfig struct {
|
||||
@@ -70,33 +83,24 @@ func registryMatchDomain(reg Registry) string {
|
||||
}
|
||||
|
||||
func (c *Config) PreProcess() (*ShimAuthConfig, error) {
|
||||
if c.ImageShimSocket == "" {
|
||||
c.ImageShimSocket = SealosShimSock
|
||||
}
|
||||
logger.Info("shim-socket: %s", c.ImageShimSocket)
|
||||
logger.Info("cri-socket: %s", c.RuntimeSocket)
|
||||
logger.Info("hub-address: %s", c.Address)
|
||||
logger.Info("auth: %s", c.Auth)
|
||||
rawURL, err := url.Parse(c.Address)
|
||||
if err != nil {
|
||||
logger.Warn("url parse error: %+v", err)
|
||||
}
|
||||
domain := rawURL.Host
|
||||
if c.Timeout.Duration.Milliseconds() == 0 {
|
||||
c.Timeout = metav1.Duration{}
|
||||
c.Timeout.Duration, _ = time.ParseDuration("15m")
|
||||
}
|
||||
if c.ReloadInterval.Duration <= 0 {
|
||||
c.ReloadInterval = metav1.Duration{Duration: DefaultReloadInterval}
|
||||
}
|
||||
|
||||
logger.Info("RegistryDomain: %v", domain)
|
||||
logger.Info("Force: %v", c.Force)
|
||||
logger.Info("Debug: %v", c.Debug)
|
||||
logger.CfgConsoleLogger(c.Debug, false)
|
||||
logger.Info("Timeout: %v", c.Timeout)
|
||||
logger.Info("ReloadInterval: %v", c.ReloadInterval)
|
||||
shimAuth := new(ShimAuthConfig)
|
||||
if c.ImageShimSocket == "" {
|
||||
c.ImageShimSocket = SealosShimSock
|
||||
}
|
||||
rawURL, err := url.Parse(c.Address)
|
||||
if err != nil {
|
||||
logger.Warn("url parse error: %+v", err)
|
||||
}
|
||||
domain := rawURL.Host
|
||||
if c.Timeout.Duration.Milliseconds() == 0 {
|
||||
c.Timeout = metav1.Duration{}
|
||||
c.Timeout.Duration, _ = time.ParseDuration("15m")
|
||||
}
|
||||
if c.ReloadInterval.Duration <= 0 {
|
||||
c.ReloadInterval = metav1.Duration{Duration: DefaultReloadInterval}
|
||||
}
|
||||
logger.CfgConsoleLogger(c.Debug, false)
|
||||
c.Cache.normalize()
|
||||
shimAuth := new(ShimAuthConfig)
|
||||
|
||||
splitNameAndPasswd := func(auth string) (string, string) {
|
||||
var username, password string
|
||||
@@ -138,8 +142,8 @@ func (c *Config) PreProcess() (*ShimAuthConfig, error) {
|
||||
}
|
||||
shimAuth.CRIConfigs = criAuth
|
||||
shimAuth.SkipLoginRegistries = skipLogin
|
||||
logger.Info("criRegistryAuth: %+v", shimAuth.CRIConfigs)
|
||||
}
|
||||
logger.Debug("criRegistryAuth: %+v", shimAuth.CRIConfigs)
|
||||
}
|
||||
|
||||
{
|
||||
offlineName, offlinePasswd := splitNameAndPasswd(c.Auth)
|
||||
@@ -149,12 +153,12 @@ func (c *Config) PreProcess() (*ShimAuthConfig, error) {
|
||||
Password: offlinePasswd,
|
||||
ServerAddress: c.Address,
|
||||
}}
|
||||
logger.Info("criOfflineAuth: %+v", shimAuth.OfflineCRIConfigs)
|
||||
}
|
||||
logger.Debug("criOfflineAuth: %+v", shimAuth.OfflineCRIConfigs)
|
||||
}
|
||||
|
||||
if c.Address == "" {
|
||||
return nil, errors.New("registry addr is empty")
|
||||
}
|
||||
if c.Address == "" {
|
||||
return nil, errors.New("registry addr is empty")
|
||||
}
|
||||
if c.RuntimeSocket == "" {
|
||||
socket, err := cri.DetectCRISocket()
|
||||
if err != nil {
|
||||
@@ -162,12 +166,36 @@ func (c *Config) PreProcess() (*ShimAuthConfig, error) {
|
||||
}
|
||||
c.RuntimeSocket = socket
|
||||
}
|
||||
if !c.Force {
|
||||
if !fileutil.IsExist(c.RuntimeSocket) {
|
||||
return nil, errors.New("cri is running?")
|
||||
}
|
||||
if !c.Force {
|
||||
if !fileutil.IsExist(c.RuntimeSocket) {
|
||||
return nil, errors.New("cri is running?")
|
||||
}
|
||||
}
|
||||
return shimAuth, nil
|
||||
}
|
||||
|
||||
func (c *CacheConfig) normalize() {
|
||||
if c.ImageCacheSize == 0 {
|
||||
c.ImageCacheSize = defaultImageCacheSize
|
||||
}
|
||||
if c.ImageCacheSize < 0 {
|
||||
logger.Warn("received negative cache size %d, disabling cache", c.ImageCacheSize)
|
||||
c.ImageCacheSize = 0
|
||||
}
|
||||
|
||||
if c.ImageCacheTTL.Duration <= 0 {
|
||||
c.ImageCacheTTL = metav1.Duration{Duration: defaultImageCacheTTL}
|
||||
}
|
||||
if c.DomainCacheTTL.Duration <= 0 {
|
||||
c.DomainCacheTTL = metav1.Duration{Duration: defaultDomainCacheTTL}
|
||||
}
|
||||
if c.DisableStats {
|
||||
c.StatsLogInterval = metav1.Duration{}
|
||||
return
|
||||
}
|
||||
if c.StatsLogInterval.Duration <= 0 {
|
||||
c.StatsLogInterval = metav1.Duration{Duration: defaultCacheStatsInterval}
|
||||
}
|
||||
return shimAuth, nil
|
||||
}
|
||||
|
||||
func Unmarshal(path string) (*Config, error) {
|
||||
@@ -179,7 +207,11 @@ func Unmarshal(path string) (*Config, error) {
|
||||
}
|
||||
|
||||
func UnmarshalData(metadata []byte) (*Config, error) {
|
||||
cfg := &Config{}
|
||||
cfg := &Config{
|
||||
Cache: CacheConfig{
|
||||
StatsLogInterval: metav1.Duration{Duration: defaultCacheStatsInterval},
|
||||
},
|
||||
}
|
||||
if err := yaml.Unmarshal(metadata, cfg); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -16,7 +16,11 @@ limitations under the License.
|
||||
|
||||
package types
|
||||
|
||||
import "testing"
|
||||
import (
|
||||
"testing"
|
||||
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
)
|
||||
|
||||
func TestUnmarshal(t *testing.T) {
|
||||
cfg, err := Unmarshal("testdata/image-cri-shim.yaml")
|
||||
@@ -78,3 +82,61 @@ func TestInlineRegistriesPreProcess(t *testing.T) {
|
||||
t.Fatalf("expected %s to skip login, got %#v", publicDomain, auth.SkipLoginRegistries)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCacheConfigNormalize(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
in CacheConfig
|
||||
expected CacheConfig
|
||||
}{
|
||||
{
|
||||
name: "defaults applied",
|
||||
in: CacheConfig{},
|
||||
expected: CacheConfig{
|
||||
ImageCacheSize: defaultImageCacheSize,
|
||||
ImageCacheTTL: metav1.Duration{Duration: defaultImageCacheTTL},
|
||||
DomainCacheTTL: metav1.Duration{Duration: defaultDomainCacheTTL},
|
||||
StatsLogInterval: metav1.Duration{Duration: defaultCacheStatsInterval},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "disable stats respected",
|
||||
in: CacheConfig{
|
||||
ImageCacheSize: 512,
|
||||
DisableStats: true,
|
||||
},
|
||||
expected: CacheConfig{
|
||||
ImageCacheSize: 512,
|
||||
ImageCacheTTL: metav1.Duration{Duration: defaultImageCacheTTL},
|
||||
DomainCacheTTL: metav1.Duration{Duration: defaultDomainCacheTTL},
|
||||
DisableStats: true,
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "negative size disables cache",
|
||||
in: CacheConfig{
|
||||
ImageCacheSize: -10,
|
||||
},
|
||||
expected: CacheConfig{
|
||||
ImageCacheSize: 0,
|
||||
ImageCacheTTL: metav1.Duration{Duration: defaultImageCacheTTL},
|
||||
DomainCacheTTL: metav1.Duration{Duration: defaultDomainCacheTTL},
|
||||
StatsLogInterval: metav1.Duration{Duration: defaultCacheStatsInterval},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range cases {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
cfg := tt.in
|
||||
cfg.normalize()
|
||||
if cfg.ImageCacheSize != tt.expected.ImageCacheSize ||
|
||||
cfg.ImageCacheTTL != tt.expected.ImageCacheTTL ||
|
||||
cfg.DomainCacheTTL != tt.expected.DomainCacheTTL ||
|
||||
cfg.DisableStats != tt.expected.DisableStats ||
|
||||
cfg.StatsLogInterval != tt.expected.StatsLogInterval {
|
||||
t.Fatalf("expected %+v, got %+v", tt.expected, cfg)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
+52
-5
@@ -51,6 +51,7 @@ type registryConfigSpec struct {
|
||||
Force *bool `yaml:"force"`
|
||||
Debug *bool `yaml:"debug"`
|
||||
Timeout string `yaml:"timeout"`
|
||||
Cache *cacheSpec `yaml:"cache"`
|
||||
}
|
||||
|
||||
type sealedConfig struct {
|
||||
@@ -68,6 +69,14 @@ type registryAuth struct {
|
||||
Password string `yaml:"password"`
|
||||
}
|
||||
|
||||
type cacheSpec struct {
|
||||
ImageCacheSize *int `yaml:"imageCacheSize"`
|
||||
ImageCacheTTL string `yaml:"imageCacheTTL"`
|
||||
DomainCacheTTL string `yaml:"domainCacheTTL"`
|
||||
StatsLogInterval string `yaml:"statsLogInterval"`
|
||||
DisableStats *bool `yaml:"disableStats"`
|
||||
}
|
||||
|
||||
// SyncConfigFromConfigMap reads the kube-system/image-cri-shim ConfigMap and, when available,
|
||||
// synchronizes the registries.yaml content into the local shim configuration file. If the cluster
|
||||
// or ConfigMap cannot be reached, the function simply skips the update.
|
||||
@@ -90,11 +99,11 @@ func SyncConfigFromConfigMap(ctx context.Context, configPath string) {
|
||||
logger.Debug("failed to read ConfigMap %s/%s: %v", shimConfigMapNamespace, shimConfigMapName, err)
|
||||
return
|
||||
}
|
||||
if !applyConfigMapToFile(configPath, cm) {
|
||||
logger.Debug("ConfigMap %s/%s produced no updates", shimConfigMapNamespace, shimConfigMapName)
|
||||
return
|
||||
}
|
||||
logger.Info("syncing image-cri-shim config from ConfigMap completed")
|
||||
if !applyConfigMapToFile(configPath, cm) {
|
||||
logger.Debug("ConfigMap %s/%s produced no updates", shimConfigMapNamespace, shimConfigMapName)
|
||||
return
|
||||
}
|
||||
logger.Debug("syncing image-cri-shim config from ConfigMap completed")
|
||||
}
|
||||
|
||||
func buildKubeClient() (kubernetes.Interface, error) {
|
||||
@@ -220,6 +229,7 @@ func mergeShimConfig(cfg *Config, spec *registryConfigSpec) {
|
||||
if cfg.ReloadInterval.Duration <= 0 {
|
||||
cfg.ReloadInterval.Duration = DefaultReloadInterval
|
||||
}
|
||||
applyCacheSpec(cfg, spec.Cache)
|
||||
}
|
||||
|
||||
func buildAuth(username, password string) string {
|
||||
@@ -230,3 +240,40 @@ func buildAuth(username, password string) string {
|
||||
}
|
||||
return user + ":" + pass
|
||||
}
|
||||
|
||||
func applyCacheSpec(cfg *Config, spec *cacheSpec) {
|
||||
if cfg == nil || spec == nil {
|
||||
return
|
||||
}
|
||||
if spec.ImageCacheSize != nil {
|
||||
cfg.Cache.ImageCacheSize = *spec.ImageCacheSize
|
||||
}
|
||||
if disable := spec.DisableStats; disable != nil {
|
||||
cfg.Cache.DisableStats = *disable
|
||||
}
|
||||
if ttl := strings.TrimSpace(spec.ImageCacheTTL); ttl != "" {
|
||||
if dur, err := time.ParseDuration(ttl); err != nil {
|
||||
logger.Warn("failed to parse imageCacheTTL %q: %v", ttl, err)
|
||||
} else {
|
||||
cfg.Cache.ImageCacheTTL.Duration = dur
|
||||
}
|
||||
}
|
||||
if ttl := strings.TrimSpace(spec.DomainCacheTTL); ttl != "" {
|
||||
if dur, err := time.ParseDuration(ttl); err != nil {
|
||||
logger.Warn("failed to parse domainCacheTTL %q: %v", ttl, err)
|
||||
} else {
|
||||
cfg.Cache.DomainCacheTTL.Duration = dur
|
||||
}
|
||||
}
|
||||
if interval := strings.TrimSpace(spec.StatsLogInterval); interval != "" || (spec.DisableStats != nil && *spec.DisableStats) {
|
||||
if spec.DisableStats != nil && *spec.DisableStats {
|
||||
cfg.Cache.StatsLogInterval.Duration = 0
|
||||
} else if interval != "" {
|
||||
if dur, err := time.ParseDuration(interval); err != nil {
|
||||
logger.Warn("failed to parse statsLogInterval %q: %v", interval, err)
|
||||
} else {
|
||||
cfg.Cache.StatsLogInterval.Duration = dur
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+36
@@ -40,6 +40,12 @@ registries:
|
||||
username: "3"
|
||||
password: "4"
|
||||
reloadInterval: 5s
|
||||
cache:
|
||||
imageCacheSize: 2048
|
||||
imageCacheTTL: 45m
|
||||
domainCacheTTL: 15m
|
||||
statsLogInterval: 120s
|
||||
disableStats: true
|
||||
`
|
||||
|
||||
func TestMergeShimConfig(t *testing.T) {
|
||||
@@ -68,6 +74,21 @@ func TestMergeShimConfig(t *testing.T) {
|
||||
if cfg.Timeout.Duration <= 0 {
|
||||
t.Fatalf("expected timeout to be set, got %s", cfg.Timeout.Duration)
|
||||
}
|
||||
if cfg.Cache.ImageCacheSize != 2048 {
|
||||
t.Fatalf("expected cache size 2048, got %d", cfg.Cache.ImageCacheSize)
|
||||
}
|
||||
if cfg.Cache.ImageCacheTTL.Duration != 45*time.Minute {
|
||||
t.Fatalf("expected image cache ttl 45m, got %s", cfg.Cache.ImageCacheTTL.Duration)
|
||||
}
|
||||
if cfg.Cache.DomainCacheTTL.Duration != 15*time.Minute {
|
||||
t.Fatalf("expected domain cache ttl 15m, got %s", cfg.Cache.DomainCacheTTL.Duration)
|
||||
}
|
||||
if cfg.Cache.DisableStats != true {
|
||||
t.Fatalf("expected disableStats true")
|
||||
}
|
||||
if cfg.Cache.StatsLogInterval.Duration != 0 {
|
||||
t.Fatalf("expected stats interval 0 due to disableStats, got %s", cfg.Cache.StatsLogInterval.Duration)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeShimConfigKeepsDefaults(t *testing.T) {
|
||||
@@ -127,6 +148,21 @@ func TestSyncConfigFromConfigMapWritesFile(t *testing.T) {
|
||||
if merged.Timeout.Duration != 15*time.Minute {
|
||||
t.Fatalf("unexpected timeout: %s", merged.Timeout.Duration)
|
||||
}
|
||||
if merged.Cache.ImageCacheSize != 2048 {
|
||||
t.Fatalf("unexpected cache size: %d", merged.Cache.ImageCacheSize)
|
||||
}
|
||||
if merged.Cache.ImageCacheTTL.Duration != 45*time.Minute {
|
||||
t.Fatalf("unexpected cache ttl: %s", merged.Cache.ImageCacheTTL.Duration)
|
||||
}
|
||||
if merged.Cache.DomainCacheTTL.Duration != 15*time.Minute {
|
||||
t.Fatalf("unexpected domain cache ttl: %s", merged.Cache.DomainCacheTTL.Duration)
|
||||
}
|
||||
if !merged.Cache.DisableStats {
|
||||
t.Fatalf("expected stats disabled")
|
||||
}
|
||||
if merged.Cache.StatsLogInterval.Duration != 0 {
|
||||
t.Fatalf("expected stats interval 0, got %s", merged.Cache.StatsLogInterval.Duration)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncConfigFromConfigMapMissingData(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user