feat: remove containerd service dependency

README.md

@@ -59,17 +59,7 @@ $ cp misc/config/config.estargz.yaml misc/config/config.yaml
 Following comments in `misc/config/config.yaml`, ensure base64 encoded auth string and webhook auth header be configured correctly.
 
 ### Boot service
-
-1. Ensure containerd service is running
-
-Acceleration service depends on containerd service to manage image content, containerd service is used default if you have installed the latest docker.
-
-``` shell
-$ ps aux | grep containerd
-root        929  0.5  0.5 2222284 22368 ?       Ssl  Sep16   4:27 /usr/local/bin/containerd
-```
-
-2. Boot daemon to serve webhook request in PUSH_ARTIFACT event
+Boot daemon to serve webhook request in PUSH_ARTIFACT event
 
 ``` shell
 $ ./acceld --config ./misc/config/config.yaml
@@ -111,7 +101,7 @@ After that, we will find converted artifact in Harbor interface with name `<harb
 
 ### One-time mode
 
-One-time mode allows to do a conversion without starting the `acceld` service (still requires a local containerd service), using `accelctl` like this:
+One-time mode allows to do a conversion without starting the `acceld` service, using `accelctl` like this:
 
 ```
 $ accelctl convert --config ./misc/config/config.yaml 192.168.1.1/library/nginx:latest

go.mod

@@ -19,6 +19,7 @@ require (
 	github.com/sirupsen/logrus v1.9.0
 	github.com/stretchr/testify v1.8.2
 	github.com/urfave/cli/v2 v2.25.0
+	golang.org/x/sync v0.1.0
 	gopkg.in/yaml.v3 v3.0.1
 )
 
@@ -95,7 +96,6 @@ require (
 	golang.org/x/crypto v0.6.0 // indirect
 	golang.org/x/mod v0.9.0 // indirect
 	golang.org/x/net v0.8.0 // indirect
-	golang.org/x/sync v0.1.0 // indirect
 	golang.org/x/sys v0.6.0 // indirect
 	golang.org/x/text v0.8.0 // indirect
 	golang.org/x/time v0.3.0 // indirect

misc/config/config.estargz.yaml

@@ -26,10 +26,7 @@ provider:
         auth_header: header
     localhost:
       auth: YWRtaW46SGFyYm9yMTIzNDU=
-  containerd:
-    # ensure containerd service listening on this address
-    address: /run/containerd/containerd.sock
-    snapshotter: overlayfs
+  work_dir: /tmp
 
 converter:
   # number of worker for executing conversion task

misc/config/config.nydus.ref.yaml

@@ -26,10 +26,7 @@ provider:
         auth_header: header
     localhost:
       auth: YWRtaW46SGFyYm9yMTIzNDU=
-  containerd:
-    # ensure containerd service listening on this address
-    address: /run/containerd/containerd.sock
-    snapshotter: overlayfs
+  work_dir: /tmp
 
 converter:
   # number of worker for executing conversion task

misc/config/config.nydus.yaml

@@ -26,10 +26,7 @@ provider:
         auth_header: header
     localhost:
       auth: YWRtaW46SGFyYm9yMTIzNDU=
-  containerd:
-    # ensure containerd service listening on this address
-    address: /run/containerd/containerd.sock
-    snapshotter: overlayfs
+  work_dir: /tmp
 
 converter:
   # number of worker for executing conversion task

pkg/adapter/adapter.go

@@ -16,10 +16,8 @@ package adapter
 
 import (
 	"context"
-	"fmt"
 	"strings"
 
-	"github.com/containerd/containerd"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 
@@ -46,28 +44,19 @@ type Adapter interface {
 
 type LocalAdapter struct {
 	cfg    *config.Config
-	client *containerd.Client
 	rule   *Rule
 	worker *Worker
 	cvt    *converter.Converter
 }
 
 func NewLocalAdapter(cfg *config.Config) (*LocalAdapter, error) {
-	client, err := containerd.New(
-		cfg.Provider.Containerd.Address,
-		containerd.WithDefaultNamespace("harbor-acceleration-service"),
-	)
-	if err != nil {
-		return nil, errors.Wrap(err, "create containerd client")
-	}
-
 	allPlatforms := len(strings.TrimSpace(cfg.Converter.Platforms)) == 0
 	platformMC, err := platformutil.ParsePlatforms(allPlatforms, cfg.Converter.Platforms)
 	if err != nil {
 		return nil, errors.Wrap(err, "invalid platform configuration")
 	}
 
-	provider, err := content.NewLocalProvider(client, cfg.Host, platformMC)
+	provider, err := content.NewLocalProvider(cfg.Provider.WorkDir, cfg.Host, platformMC)
 	if err != nil {
 		return nil, errors.Wrap(err, "create content provider")
 	}
@@ -92,7 +81,6 @@ func NewLocalAdapter(cfg *config.Config) (*LocalAdapter, error) {
 
 	handler := &LocalAdapter{
 		cfg:    cfg,
-		client: client,
 		rule:   rule,
 		worker: worker,
 		cvt:    cvt,
@@ -110,12 +98,6 @@ func (adp *LocalAdapter) Convert(ctx context.Context, source string) error {
 		}
 		return errors.Wrap(err, "create target reference by rule")
 	}
-
-	ctx, done, err := adp.client.WithLease(ctx)
-	if err != nil {
-		return errors.Wrap(err, "create lease")
-	}
-	defer done(ctx)
 	_, err = adp.cvt.Convert(ctx, source, target)
 	return err
 }
@@ -145,16 +127,6 @@ func (adp *LocalAdapter) Dispatch(ctx context.Context, ref string, sync bool) er
 }
 
 func (adp *LocalAdapter) CheckHealth(ctx context.Context) error {
-	health, err := adp.client.IsServing(ctx)
-
-	msg := "containerd service is unhealthy"
-	if err != nil {
-		return errors.Wrap(err, msg)
-	}
-
-	if !health {
-		return fmt.Errorf(msg)
-	}
-
+	// TODO：return the status of boltdb
 	return nil
 }

pkg/config/config.go

@@ -46,8 +46,8 @@ type MetricConfig struct {
 }
 
 type ProviderConfig struct {
-	Source     map[string]SourceConfig `yaml:"source"`
-	Containerd ContainerdConfig        `yaml:"containerd"`
+	Source  map[string]SourceConfig `yaml:"source"`
+	WorkDir string                  `yaml:"work_dir"`
 }
 
 type Webhook struct {
@@ -60,11 +60,6 @@ type SourceConfig struct {
 	Webhook  Webhook `yaml:"webhook"`
 }
 
-type ContainerdConfig struct {
-	Address     string `yaml:"address"`
-	Snapshotter string `yaml:"snapshotter"`
-}
-
 type ConversionRule struct {
 	TagSuffix string `yaml:"tag_suffix"`
 }

pkg/content/local.go

@@ -16,34 +16,40 @@ package content
 
 import (
 	"context"
+	"sync"
 
 	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/content"
-	"github.com/containerd/containerd/images"
+	"github.com/containerd/containerd/content/local"
+	"github.com/containerd/containerd/errdefs"
 	"github.com/containerd/containerd/platforms"
 	"github.com/containerd/containerd/remotes"
 	"github.com/goharbor/acceleration-service/pkg/remote"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/pkg/errors"
-	"github.com/sirupsen/logrus"
 )
 
-var logger = logrus.WithField("module", "content")
-
 type LocalProvider struct {
+	mutex        sync.Mutex
+	images       map[string]*ocispec.Descriptor
 	usePlainHTTP bool
-	client       *containerd.Client
+	store        content.Store
 	hosts        remote.HostFunc
 	platformMC   platforms.MatchComparer
 }
 
 func NewLocalProvider(
-	client *containerd.Client,
+	workDir string,
 	hosts remote.HostFunc,
 	platformMC platforms.MatchComparer,
 ) (Provider, error) {
+	store, err := local.NewLabeledStore(workDir, newMemoryLabelStore())
+	if err != nil {
+		return nil, errors.Wrap(err, "create local provider")
+	}
 	return &LocalProvider{
-		client:     client,
+		store:      store,
+		images:     make(map[string]*ocispec.Descriptor),
 		hosts:      hosts,
 		platformMC: platformMC,
 	}, nil
@@ -67,25 +73,17 @@ func (pvd *LocalProvider) Pull(ctx context.Context, ref string) error {
 		return err
 	}
 
-	opts := []containerd.RemoteOpt{
-		// TODO: sets max concurrent downloaded layer limit by containerd.WithMaxConcurrentDownloads.
-		containerd.WithPlatformMatcher(pvd.platformMC),
-		containerd.WithImageHandler(images.HandlerFunc(
-			func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
-				if images.IsLayerType(desc.MediaType) {
-					logger.Debugf("pulling layer %s", desc.Digest)
-				}
-				return nil, nil
-			},
-		)),
-		containerd.WithResolver(resolver),
+	// TODO: sets max concurrent downloaded layer limit by containerd.WithMaxConcurrentDownloads.
+	rc := &containerd.RemoteContext{
+		Resolver:        resolver,
+		PlatformMatcher: pvd.platformMC,
 	}
 
-	// Pull the source image from remote registry.
-	_, err = pvd.client.Fetch(ctx, ref, opts...)
+	img, err := fetch(ctx, pvd.store, rc, ref, 0)
 	if err != nil {
 		return errors.Wrap(err, "pull source image")
 	}
+	pvd.setImage(ref, &img.Target)
 
 	return nil
 }
@@ -96,24 +94,33 @@ func (pvd *LocalProvider) Push(ctx context.Context, desc ocispec.Descriptor, ref
 		return err
 	}
 
-	opts := []containerd.RemoteOpt{
-		containerd.WithResolver(resolver),
-		containerd.WithPlatformMatcher(pvd.platformMC),
+	rc := &containerd.RemoteContext{
+		Resolver:        resolver,
+		PlatformMatcher: pvd.platformMC,
 	}
 
-	// TODO: sets max concurrent uploaded layer limit by containerd.WithMaxConcurrentUploadedLayers.
-	return pvd.client.Push(ctx, ref, desc, opts...)
+	return push(ctx, pvd.store, rc, desc, ref)
 }
 
 func (pvd *LocalProvider) Image(ctx context.Context, ref string) (*ocispec.Descriptor, error) {
-	image, err := pvd.client.GetImage(ctx, ref)
-	if err != nil {
-		return nil, err
-	}
-	target := containerd.NewImageWithPlatform(pvd.client, image.Metadata(), pvd.platformMC).Target()
-	return &target, nil
+	return pvd.getImage(ref)
 }
 
 func (pvd *LocalProvider) ContentStore() content.Store {
-	return pvd.client.ContentStore()
+	return pvd.store
+}
+
+func (pvd *LocalProvider) setImage(ref string, image *ocispec.Descriptor) {
+	pvd.mutex.Lock()
+	defer pvd.mutex.Unlock()
+	pvd.images[ref] = image
+}
+
+func (pvd *LocalProvider) getImage(ref string) (*ocispec.Descriptor, error) {
+	pvd.mutex.Lock()
+	defer pvd.mutex.Unlock()
+	if desc, ok := pvd.images[ref]; ok {
+		return desc, nil
+	}
+	return nil, errdefs.ErrNotFound
 }