diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
index 83d456fc3f8e3..7ba1683970a85 100644
--- a/.github/CODEOWNERS
+++ b/.github/CODEOWNERS
@@ -212,6 +212,7 @@
 /pkg/collector/corechecks/cluster/      @DataDog/container-integrations
 /pkg/collector/corechecks/cluster/orchestrator  @DataDog/container-app
 /pkg/collector/corechecks/containers/   @DataDog/container-integrations
+/pkg/collector/corechecks/containerimage/       @DataDog/container-integrations
 /pkg/collector/corechecks/containerlifecycle/   @DataDog/container-integrations
 /pkg/collector/corechecks/ebpf/         @DataDog/container-integrations
 /pkg/collector/corechecks/embed/        @Datadog/agent-platform
diff --git a/cmd/agent/subcommands/run/command.go b/cmd/agent/subcommands/run/command.go
index 07a6399a871a6..cab66563f728b 100644
--- a/cmd/agent/subcommands/run/command.go
+++ b/cmd/agent/subcommands/run/command.go
@@ -71,6 +71,7 @@ import (
 	_ "github.com/DataDog/datadog-agent/pkg/collector/corechecks/cluster/ksm"
 	_ "github.com/DataDog/datadog-agent/pkg/collector/corechecks/cluster/kubernetesapiserver"
 	_ "github.com/DataDog/datadog-agent/pkg/collector/corechecks/cluster/orchestrator"
+	_ "github.com/DataDog/datadog-agent/pkg/collector/corechecks/containerimage"
 	_ "github.com/DataDog/datadog-agent/pkg/collector/corechecks/containerlifecycle"
 	_ "github.com/DataDog/datadog-agent/pkg/collector/corechecks/containers/containerd"
 	_ "github.com/DataDog/datadog-agent/pkg/collector/corechecks/containers/cri"
@@ -374,6 +375,7 @@ func startAgent(cliParams *cliParams, flare flare.Component) error {
 	opts := aggregator.DefaultAgentDemultiplexerOptions(forwarderOpts)
 	opts.EnableNoAggregationPipeline = pkgconfig.Datadog.GetBool("dogstatsd_no_aggregation_pipeline")
 	opts.UseContainerLifecycleForwarder = pkgconfig.Datadog.GetBool("container_lifecycle.enabled")
+	opts.UseContainerImageForwarder = pkgconfig.Datadog.GetBool("container_image.enabled")
 	demux = aggregator.InitAndStartAgentDemultiplexer(opts, hostnameDetected)
 
 	// Setup stats telemetry handler
diff --git a/pkg/autodiscovery/listeners/container.go b/pkg/autodiscovery/listeners/container.go
index 9aca9e2d3caf7..cc7fdc502d20b 100644
--- a/pkg/autodiscovery/listeners/container.go
+++ b/pkg/autodiscovery/listeners/container.go
@@ -189,7 +189,7 @@ func computeContainerServiceIDs(entity string, image string, labels map[string]s
 	ids := []string{entity}
 
 	// Add Image names (long then short if different)
-	long, short, _, err := containers.SplitImageName(image)
+	long, _, short, _, err := containers.SplitImageName(image)
 	if err != nil {
 		log.Warnf("error while spliting image name: %s", err)
 	}
diff --git a/pkg/collector/corechecks/containerimage/check.go b/pkg/collector/corechecks/containerimage/check.go
new file mode 100644
index 0000000000000..15f1fdcebf464
--- /dev/null
+++ b/pkg/collector/corechecks/containerimage/check.go
@@ -0,0 +1,166 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2022-present Datadog, Inc.
+
+package containerimage
+
+import (
+	"errors"
+	"time"
+
+	yaml "gopkg.in/yaml.v2"
+
+	"github.com/DataDog/datadog-agent/pkg/autodiscovery/integration"
+	"github.com/DataDog/datadog-agent/pkg/collector/check"
+	core "github.com/DataDog/datadog-agent/pkg/collector/corechecks"
+	ddConfig "github.com/DataDog/datadog-agent/pkg/config"
+	"github.com/DataDog/datadog-agent/pkg/util/log"
+	"github.com/DataDog/datadog-agent/pkg/workloadmeta"
+)
+
+const (
+	checkName = "container_image"
+)
+
+func init() {
+	core.RegisterCheck(checkName, CheckFactory)
+}
+
+// Config holds the container_image check configuration
+type Config struct {
+	chunkSize                  int `yaml:"chunk_size"`
+	newImagesMaxLatencySeconds int `yaml:"new_images_max_latency_seconds"`
+	periodicRefreshSeconds     int `yaml:"periodic_refresh_seconds"`
+}
+
+type configValueRange struct {
+	min      int
+	max      int
+	default_ int
+}
+
+var /* const */ (
+	chunkSizeValueRange = &configValueRange{
+		min:      1,
+		max:      100,
+		default_: 10,
+	}
+
+	newImagesMaxLatencySecondsValueRange = &configValueRange{
+		min:      1,   // 1 s
+		max:      300, // 5 min
+		default_: 30,  // 30 s
+	}
+
+	periodicRefreshSecondsValueRange = &configValueRange{
+		min:      60,    // 1 min
+		max:      86400, // 1 day
+		default_: 300,   // 5 min
+	}
+)
+
+func validateValue(val *int, range_ *configValueRange) {
+	if *val == 0 {
+		*val = range_.default_
+	} else if *val < range_.min {
+		*val = range_.min
+	} else if *val > range_.max {
+		*val = range_.max
+	}
+}
+
+func (c *Config) Parse(data []byte) error {
+	if err := yaml.Unmarshal(data, c); err != nil {
+		return err
+	}
+
+	validateValue(&c.chunkSize, chunkSizeValueRange)
+	validateValue(&c.newImagesMaxLatencySeconds, newImagesMaxLatencySecondsValueRange)
+	validateValue(&c.periodicRefreshSeconds, periodicRefreshSecondsValueRange)
+
+	return nil
+}
+
+// Check reports container images
+type Check struct {
+	core.CheckBase
+	workloadmetaStore workloadmeta.Store
+	instance          *Config
+	processor         *processor
+	stopCh            chan struct{}
+}
+
+// CheckFactory registers the container_image check
+func CheckFactory() check.Check {
+	return &Check{
+		CheckBase:         core.NewCheckBase(checkName),
+		workloadmetaStore: workloadmeta.GetGlobalStore(),
+		instance:          &Config{},
+		stopCh:            make(chan struct{}),
+	}
+}
+
+// Configure parses the check configuration and initializes the container_image check
+func (c *Check) Configure(integrationConfigDigest uint64, config, initConfig integration.Data, source string) error {
+	if !ddConfig.Datadog.GetBool("container_image.enabled") {
+		return errors.New("collection of container images is disabled")
+	}
+
+	if err := c.CommonConfigure(integrationConfigDigest, initConfig, config, source); err != nil {
+		return err
+	}
+
+	if err := c.instance.Parse(config); err != nil {
+		return err
+	}
+
+	sender, err := c.GetSender()
+	if err != nil {
+		return err
+	}
+
+	c.processor = newProcessor(sender, c.instance.chunkSize, time.Duration(c.instance.newImagesMaxLatencySeconds)*time.Second)
+
+	return nil
+}
+
+// Run starts the container_image check
+func (c *Check) Run() error {
+	log.Infof("Starting long-running check %q", c.ID())
+	defer log.Infof("Shutting down long-running check %q", c.ID())
+
+	imgEventsCh := c.workloadmetaStore.Subscribe(
+		checkName,
+		workloadmeta.NormalPriority,
+		workloadmeta.NewFilter(
+			[]workloadmeta.Kind{workloadmeta.KindContainerImageMetadata},
+			workloadmeta.SourceAll,
+			workloadmeta.EventTypeSet, // We don’t care about images removal because we just have to wait for them to expire on BE side once we stopped refreshing them periodically.
+		),
+	)
+
+	imgRefreshTicker := time.NewTicker(time.Duration(c.instance.periodicRefreshSeconds) * time.Second)
+
+	for {
+		select {
+		case eventBundle := <-imgEventsCh:
+			c.processor.processEvents(eventBundle)
+		case <-imgRefreshTicker.C:
+			c.processor.processRefresh(c.workloadmetaStore.ListImages())
+		case <-c.stopCh:
+			c.processor.stop()
+			return nil
+		}
+	}
+}
+
+// Stop stops the container_image check
+func (c *Check) Stop() {
+	close(c.stopCh)
+}
+
+// Interval returns 0. It makes container_image a long-running check
+func (c *Check) Interval() time.Duration {
+	return 0
+}
diff --git a/pkg/collector/corechecks/containerimage/processor.go b/pkg/collector/corechecks/containerimage/processor.go
new file mode 100644
index 0000000000000..99e9a58693000
--- /dev/null
+++ b/pkg/collector/corechecks/containerimage/processor.go
@@ -0,0 +1,84 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2022-present Datadog, Inc.
+
+package containerimage
+
+import (
+	"time"
+
+	"github.com/DataDog/datadog-agent/pkg/aggregator"
+	"github.com/DataDog/datadog-agent/pkg/util/log"
+	"github.com/DataDog/datadog-agent/pkg/workloadmeta"
+
+	"github.com/DataDog/agent-payload/v5/contimage"
+	model "github.com/DataDog/agent-payload/v5/contimage"
+)
+
+type processor struct {
+	queue chan *model.ContainerImage
+}
+
+func newProcessor(sender aggregator.Sender, maxNbItem int, maxRetentionTime time.Duration) *processor {
+	return &processor{
+		queue: newQueue(maxNbItem, maxRetentionTime, func(images []*model.ContainerImage) {
+			sender.ContainerImage([]contimage.ContainerImagePayload{
+				{
+					Version: "v1",
+					Images:  images,
+				},
+			})
+		}),
+	}
+}
+
+func (p *processor) processEvents(evBundle workloadmeta.EventBundle) {
+	close(evBundle.Ch)
+
+	log.Tracef("Processing %d events", len(evBundle.Events))
+
+	for _, event := range evBundle.Events {
+		p.processImage(event.Entity.(*workloadmeta.ContainerImageMetadata))
+	}
+}
+
+func (p *processor) processRefresh(allImages []*workloadmeta.ContainerImageMetadata) {
+	// So far, the check is refreshing all the images every 5 minutes all together.
+	for _, img := range allImages {
+		p.processImage(img)
+	}
+}
+
+func (p *processor) processImage(img *workloadmeta.ContainerImageMetadata) {
+	layers := make([]*model.ContainerImage_ContainerImageLayer, 0, len(img.Layers))
+	for _, layer := range img.Layers {
+		layers = append(layers, &model.ContainerImage_ContainerImageLayer{
+			Urls:      layer.URLs,
+			MediaType: layer.MediaType,
+			Digest:    layer.Digest,
+			Size:      layer.SizeBytes,
+		})
+	}
+
+	p.queue <- &model.ContainerImage{
+		Id:          img.ID,
+		Name:        img.Name,
+		Registry:    img.Registry,
+		ShortName:   img.ShortName,
+		Tags:        img.RepoTags,
+		Digest:      img.ID,
+		Size:        img.SizeBytes,
+		RepoDigests: img.RepoDigests,
+		Os: &model.ContainerImage_OperatingSystem{
+			Name:         img.OS,
+			Version:      img.OSVersion,
+			Architecture: img.Architecture,
+		},
+		Layers: layers,
+	}
+}
+
+func (p *processor) stop() {
+	close(p.queue)
+}
diff --git a/pkg/collector/corechecks/containerimage/processor_test.go b/pkg/collector/corechecks/containerimage/processor_test.go
new file mode 100644
index 0000000000000..771956ed9f6a6
--- /dev/null
+++ b/pkg/collector/corechecks/containerimage/processor_test.go
@@ -0,0 +1,166 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2022-present Datadog, Inc.
+
+package containerimage
+
+import (
+	"fmt"
+	"strconv"
+	"testing"
+	"time"
+
+	model "github.com/DataDog/agent-payload/v5/contimage"
+	"github.com/DataDog/datadog-agent/pkg/aggregator/mocksender"
+	"github.com/DataDog/datadog-agent/pkg/collector/check"
+	"github.com/DataDog/datadog-agent/pkg/workloadmeta"
+	"github.com/stretchr/testify/mock"
+)
+
+func TestProcessEvents(t *testing.T) {
+	sender := mocksender.NewMockSender(check.ID(""))
+	sender.On("ContainerImage", mock.Anything, mock.Anything).Return()
+	p := newProcessor(sender, 2, 50*time.Millisecond)
+
+	for i := 0; i < 3; i++ {
+		p.processEvents(workloadmeta.EventBundle{
+			Events: []workloadmeta.Event{
+				{
+					Type: workloadmeta.EventTypeSet,
+					Entity: &workloadmeta.ContainerImageMetadata{
+						EntityID: workloadmeta.EntityID{
+							Kind: workloadmeta.KindContainerImageMetadata,
+							ID:   strconv.Itoa(i),
+						},
+						ShortName:    fmt.Sprintf("short_name_%d", i),
+						RepoTags:     []string{"tag_1", "tag_2"},
+						RepoDigests:  []string{fmt.Sprintf("digest_%d", i)},
+						SizeBytes:    42,
+						OS:           "DOS",
+						OSVersion:    "6.22",
+						Architecture: "80486DX",
+						Layers: []workloadmeta.ContainerImageLayer{
+							{
+								MediaType: "media",
+								Digest:    fmt.Sprintf("digest_layer_1_%d", i),
+								SizeBytes: 43,
+								URLs:      []string{"url"},
+							},
+							{
+								MediaType: "media",
+								Digest:    fmt.Sprintf("digest_layer_2_%d", i),
+								URLs:      []string{"url"},
+								SizeBytes: 44,
+							},
+						},
+					},
+				},
+			},
+			Ch: make(chan struct{}),
+		})
+	}
+
+	sender.AssertNumberOfCalls(t, "ContainerImage", 1)
+	sender.AssertContainerImage(t, []model.ContainerImagePayload{
+		{
+			Version: "v1",
+			Images: []*model.ContainerImage{
+				{
+					Id:          "0",
+					ShortName:   "short_name_0",
+					Tags:        []string{"tag_1", "tag_2"},
+					Digest:      "0",
+					Size:        42,
+					RepoDigests: []string{"digest_0"},
+					Os: &model.ContainerImage_OperatingSystem{
+						Name:         "DOS",
+						Version:      "6.22",
+						Architecture: "80486DX",
+					},
+					Layers: []*model.ContainerImage_ContainerImageLayer{
+						{
+							Urls:      []string{"url"},
+							MediaType: "media",
+							Digest:    "digest_layer_1_0",
+							Size:      43,
+						},
+						{
+							Urls:      []string{"url"},
+							MediaType: "media",
+							Digest:    "digest_layer_2_0",
+							Size:      44,
+						},
+					},
+				},
+				{
+					Id:          "1",
+					ShortName:   "short_name_1",
+					Tags:        []string{"tag_1", "tag_2"},
+					Digest:      "1",
+					Size:        42,
+					RepoDigests: []string{"digest_1"},
+					Os: &model.ContainerImage_OperatingSystem{
+						Name:         "DOS",
+						Version:      "6.22",
+						Architecture: "80486DX",
+					},
+					Layers: []*model.ContainerImage_ContainerImageLayer{
+						{
+							Urls:      []string{"url"},
+							MediaType: "media",
+							Digest:    "digest_layer_1_1",
+							Size:      43,
+						},
+						{
+							Urls:      []string{"url"},
+							MediaType: "media",
+							Digest:    "digest_layer_2_1",
+							Size:      44,
+						},
+					},
+				},
+			},
+		},
+	})
+
+	time.Sleep(100 * time.Millisecond)
+
+	sender.AssertNumberOfCalls(t, "ContainerImage", 2)
+	sender.AssertContainerImage(t, []model.ContainerImagePayload{
+		{
+			Version: "v1",
+			Images: []*model.ContainerImage{
+				{
+					Id:          "2",
+					ShortName:   "short_name_2",
+					Tags:        []string{"tag_1", "tag_2"},
+					Digest:      "2",
+					Size:        42,
+					RepoDigests: []string{"digest_2"},
+					Os: &model.ContainerImage_OperatingSystem{
+						Name:         "DOS",
+						Version:      "6.22",
+						Architecture: "80486DX",
+					},
+					Layers: []*model.ContainerImage_ContainerImageLayer{
+						{
+							Urls:      []string{"url"},
+							MediaType: "media",
+							Digest:    "digest_layer_1_2",
+							Size:      43,
+						},
+						{
+							Urls:      []string{"url"},
+							MediaType: "media",
+							Digest:    "digest_layer_2_2",
+							Size:      44,
+						},
+					},
+				},
+			},
+		},
+	})
+
+	p.stop()
+}
diff --git a/pkg/collector/corechecks/containerimage/queue.go b/pkg/collector/corechecks/containerimage/queue.go
new file mode 100644
index 0000000000000..cfb498badd236
--- /dev/null
+++ b/pkg/collector/corechecks/containerimage/queue.go
@@ -0,0 +1,71 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2022-present Datadog, Inc.
+
+package containerimage
+
+import (
+	"time"
+
+	model "github.com/DataDog/agent-payload/v5/contimage"
+)
+
+type queue struct {
+	maxNbItem        int
+	maxRetentionTime time.Duration
+	flushCB          func([]*model.ContainerImage)
+	enqueueCh        chan *model.ContainerImage
+	data             []*model.ContainerImage
+	timer            *time.Timer
+}
+
+// newQueue returns a chan to enqueue newly discovered container images
+func newQueue(maxNbItem int, maxRetentionTime time.Duration, flushCB func([]*model.ContainerImage)) chan *model.ContainerImage {
+	q := queue{
+		maxNbItem:        maxNbItem,
+		maxRetentionTime: maxRetentionTime,
+		flushCB:          flushCB,
+		enqueueCh:        make(chan *model.ContainerImage),
+		data:             make([]*model.ContainerImage, 0, maxNbItem),
+		timer:            time.NewTimer(maxRetentionTime),
+	}
+
+	if !q.timer.Stop() {
+		<-q.timer.C
+	}
+
+	go func() {
+		for {
+			select {
+			case <-q.timer.C:
+				q.flush()
+			case img, more := <-q.enqueueCh:
+				if !more {
+					return
+				}
+				q.enqueue(img)
+			}
+		}
+	}()
+
+	return q.enqueueCh
+}
+
+func (q *queue) enqueue(elem *model.ContainerImage) {
+	if len(q.data) == 0 {
+		q.timer.Reset(q.maxRetentionTime)
+	}
+
+	q.data = append(q.data, elem)
+
+	if len(q.data) == q.maxNbItem {
+		q.flush()
+	}
+}
+
+func (q *queue) flush() {
+	q.timer.Stop()
+	q.flushCB(q.data)
+	q.data = make([]*model.ContainerImage, 0, q.maxNbItem)
+}
diff --git a/pkg/collector/corechecks/containerimage/queue_test.go b/pkg/collector/corechecks/containerimage/queue_test.go
new file mode 100644
index 0000000000000..06c88ca789ee5
--- /dev/null
+++ b/pkg/collector/corechecks/containerimage/queue_test.go
@@ -0,0 +1,71 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2022-present Datadog, Inc.
+
+package containerimage
+
+import (
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	model "github.com/DataDog/agent-payload/v5/contimage"
+	"github.com/stretchr/testify/assert"
+)
+
+func newMockFlush() (callback func([]*model.ContainerImage), getAccumulator func() [][]*model.ContainerImage) {
+	accumulator := [][]*model.ContainerImage{}
+	var mutex sync.RWMutex
+
+	callback = func(images []*model.ContainerImage) {
+		mutex.Lock()
+		defer mutex.Unlock()
+		accumulator = append(accumulator, images)
+	}
+
+	getAccumulator = func() [][]*model.ContainerImage {
+		mutex.RLock()
+		defer mutex.RUnlock()
+		return accumulator
+	}
+
+	return
+}
+
+func TestQueue(t *testing.T) {
+	callback, accumulator := newMockFlush()
+	queue := newQueue(3, 50*time.Millisecond, callback)
+
+	for i := 0; i <= 10; i++ {
+		queue <- &model.ContainerImage{
+			Id: strconv.Itoa(i),
+		}
+	}
+
+	assert.Equal(
+		t,
+		accumulator(),
+		[][]*model.ContainerImage{
+			{{Id: "0"}, {Id: "1"}, {Id: "2"}},
+			{{Id: "3"}, {Id: "4"}, {Id: "5"}},
+			{{Id: "6"}, {Id: "7"}, {Id: "8"}},
+		},
+	)
+
+	time.Sleep(100 * time.Millisecond)
+
+	assert.Equal(
+		t,
+		accumulator(),
+		[][]*model.ContainerImage{
+			{{Id: "0"}, {Id: "1"}, {Id: "2"}},
+			{{Id: "3"}, {Id: "4"}, {Id: "5"}},
+			{{Id: "6"}, {Id: "7"}, {Id: "8"}},
+			{{Id: "9"}, {Id: "10"}},
+		},
+	)
+
+	close(queue)
+}
diff --git a/pkg/collector/corechecks/containers/containerd/utils.go b/pkg/collector/corechecks/containers/containerd/utils.go
index 13f385bc0d838..9bd5c48863dc4 100644
--- a/pkg/collector/corechecks/containers/containerd/utils.go
+++ b/pkg/collector/corechecks/containers/containerd/utils.go
@@ -27,7 +27,7 @@ func getProcessorFilter(legacyFilter *containers.Filter) generic.ContainerFilter
 }
 
 func getImageTags(imageName string) []string {
-	long, short, tag, err := containers.SplitImageName(imageName)
+	long, _, short, tag, err := containers.SplitImageName(imageName)
 	if err != nil {
 		return []string{fmt.Sprintf("image:%s", imageName)}
 	}
diff --git a/pkg/collector/corechecks/containers/docker/utils.go b/pkg/collector/corechecks/containers/docker/utils.go
index b5a449dde440b..49d76d2f77923 100644
--- a/pkg/collector/corechecks/containers/docker/utils.go
+++ b/pkg/collector/corechecks/containers/docker/utils.go
@@ -42,7 +42,7 @@ func getImageTagsFromContainer(taggerEntityID string, resolvedImageName string,
 }
 
 func getImageTags(imageName string) ([]string, error) {
-	long, short, tag, err := containers.SplitImageName(imageName)
+	long, _, short, tag, err := containers.SplitImageName(imageName)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/util/containers/image.go b/pkg/util/containers/image.go
index 1959ef95720b3..8075accfe5670 100644
--- a/pkg/util/containers/image.go
+++ b/pkg/util/containers/image.go
@@ -19,16 +19,17 @@ var (
 
 // SplitImageName splits a valid image name (from ResolveImageName) and returns:
 //    - the "long image name" with registry and prefix, without tag
+//    - the registry
 //    - the "short image name", without registry, prefix nor tag
 //    - the image tag if present
 //    - an error if parsing failed
-func SplitImageName(image string) (string, string, string, error) {
+func SplitImageName(image string) (string, string, string, string, error) {
 	// See TestSplitImageName for supported formats (number 6 will surprise you!)
 	if image == "" {
-		return "", "", "", ErrEmptyImage
+		return "", "", "", "", ErrEmptyImage
 	}
 	if strings.HasPrefix(image, "sha256:") {
-		return "", "", "", ErrImageIsSha256
+		return "", "", "", "", ErrImageIsSha256
 	}
 	long := image
 	if pos := strings.LastIndex(long, "@sha"); pos > 0 {
@@ -36,9 +37,10 @@ func SplitImageName(image string) (string, string, string, error) {
 		long = long[0:pos]
 	}
 
-	var short, tag string
+	var registry, short, tag string
 	lastColon := strings.LastIndex(long, ":")
 	lastSlash := strings.LastIndex(long, "/")
+	firstSlash := strings.Index(long, "/")
 
 	if lastColon > -1 && lastColon > lastSlash {
 		// We have a tag
@@ -51,5 +53,9 @@ func SplitImageName(image string) (string, string, string, error) {
 	} else {
 		short = long
 	}
-	return long, short, tag, nil
+	if firstSlash > -1 && firstSlash != lastSlash {
+		// we have a registry
+		registry = long[:firstSlash]
+	}
+	return long, registry, short, tag, nil
 }
diff --git a/pkg/util/containers/image_test.go b/pkg/util/containers/image_test.go
index c2c541bb4b94e..7574af49675f0 100644
--- a/pkg/util/containers/image_test.go
+++ b/pkg/util/containers/image_test.go
@@ -16,43 +16,45 @@ func TestSplitImageName(t *testing.T) {
 	for nb, tc := range []struct {
 		source    string
 		longName  string
+		registry  string
 		shortName string
 		tag       string
 		err       error
 	}{
 		// Empty
-		{"", "", "", "", fmt.Errorf("empty image name")},
+		{"", "", "", "", "", fmt.Errorf("empty image name")},
 		// A sha256 string
-		{"sha256:5bef08742407efd622d243692b79ba0055383bbce12900324f75e56f589aedb0", "", "", "", fmt.Errorf("invalid image name (is a sha256)")},
+		{"sha256:5bef08742407efd622d243692b79ba0055383bbce12900324f75e56f589aedb0", "", "", "", "", fmt.Errorf("invalid image name (is a sha256)")},
 		// Shortest possibility
-		{"alpine", "alpine", "alpine", "", nil},
+		{"alpine", "alpine", "", "alpine", "", nil},
 		// Historical docker format
-		{"nginx:latest", "nginx", "nginx", "latest", nil},
+		{"nginx:latest", "nginx", "", "nginx", "latest", nil},
 		// Org prefix to be removed for short name
 		{"datadog/docker-dd-agent:latest-jmx",
-			"datadog/docker-dd-agent", "docker-dd-agent", "latest-jmx", nil},
+			"datadog/docker-dd-agent", "", "docker-dd-agent", "latest-jmx", nil},
 		// Sha-pinning used by many orchestrators -> empty tag
 		// We should not have this string here as ResolveImageName should
 		// have handled that before, but let's keep it just in case
 		{"redis@sha256:5bef08742407efd622d243692b79ba0055383bbce12900324f75e56f589aedb0",
-			"redis", "redis", "", nil},
+			"redis", "", "redis", "", nil},
 		// Quirky pinning used by swarm
 		{"org/redis:latest@sha256:5bef08742407efd622d243692b79ba0055383bbce12900324f75e56f589aedb0",
-			"org/redis", "redis", "latest", nil},
+			"org/redis", "", "redis", "latest", nil},
 		// Custom registry, simple form
 		{"myregistry.local:5000/testing/test-image:version",
-			"myregistry.local:5000/testing/test-image", "test-image", "version", nil},
+			"myregistry.local:5000/testing/test-image", "myregistry.local:5000", "test-image", "version", nil},
 		// Custom registry, most insane form possible
 		{"myregistry.local:5000/testing/test-image:version@sha256:5bef08742407efd622d243692b79ba0055383bbce12900324f75e56f589aedb0",
-			"myregistry.local:5000/testing/test-image", "test-image", "version", nil},
+			"myregistry.local:5000/testing/test-image", "myregistry.local:5000", "test-image", "version", nil},
 		// Test swarm image
 		{"dockercloud/haproxy:1.6.7@sha256:8c4ed4049f55de49cbc8d03d057a5a7e8d609c264bb75b59a04470db1d1c5121",
-			"dockercloud/haproxy", "haproxy", "1.6.7", nil},
+			"dockercloud/haproxy", "", "haproxy", "1.6.7", nil},
 	} {
 		t.Run(fmt.Sprintf("case %d: %s", nb, tc.source), func(t *testing.T) {
 			assert := assert.New(t)
-			long, short, tag, err := SplitImageName(tc.source)
+			long, registry, short, tag, err := SplitImageName(tc.source)
 			assert.Equal(tc.longName, long)
+			assert.Equal(tc.registry, registry)
 			assert.Equal(tc.shortName, short)
 			assert.Equal(tc.tag, tag)
 
diff --git a/pkg/workloadmeta/collectors/internal/containerd/image.go b/pkg/workloadmeta/collectors/internal/containerd/image.go
index 647812a8adf50..5971f86c8b999 100644
--- a/pkg/workloadmeta/collectors/internal/containerd/image.go
+++ b/pkg/workloadmeta/collectors/internal/containerd/image.go
@@ -168,12 +168,14 @@ func (c *collector) notifyEventForImage(ctx context.Context, namespace string, i
 	}
 
 	imageName := img.Name()
+	registry := ""
 	shortName := ""
 	parsedImg, err := workloadmeta.NewContainerImage(imageName)
 	if err == nil {
 		// Don't set a short name. We know that some images handled here contain
 		// "sha256" in the name, and those don't have a short name.
 	} else {
+		registry = parsedImg.Registry
 		shortName = parsedImg.ShortName
 	}
 
@@ -242,6 +244,7 @@ func (c *collector) notifyEventForImage(ctx context.Context, namespace string, i
 			Namespace: namespace,
 			Labels:    img.Labels(),
 		},
+		Registry:     registry,
 		ShortName:    shortName,
 		RepoTags:     c.repoTags[imageID],
 		RepoDigests:  repoDigests,
diff --git a/pkg/workloadmeta/collectors/internal/docker/docker.go b/pkg/workloadmeta/collectors/internal/docker/docker.go
index d2ff5092e6754..8f734c1a2e32b 100644
--- a/pkg/workloadmeta/collectors/internal/docker/docker.go
+++ b/pkg/workloadmeta/collectors/internal/docker/docker.go
@@ -278,13 +278,14 @@ func extractImage(ctx context.Context, container types.ContainerJSON, resolve re
 
 	var (
 		name      string
+		registry  string
 		shortName string
 		tag       string
 		err       error
 	)
 
 	if strings.Contains(imageSpec, "@sha256") {
-		name, shortName, tag, err = containers.SplitImageName(imageSpec)
+		name, registry, shortName, tag, err = containers.SplitImageName(imageSpec)
 		if err != nil {
 			log.Debugf("cannot split image name %q for container %q: %s", imageSpec, container.ID, err)
 		}
@@ -297,13 +298,13 @@ func extractImage(ctx context.Context, container types.ContainerJSON, resolve re
 			return image
 		}
 
-		name, shortName, tag, err = containers.SplitImageName(resolvedImageSpec)
+		name, registry, shortName, tag, err = containers.SplitImageName(resolvedImageSpec)
 		if err != nil {
 			log.Debugf("cannot split image name %q for container %q: %s", resolvedImageSpec, container.ID, err)
 
 			// fallback and try to parse the original imageSpec anyway
 			if err == containers.ErrImageIsSha256 {
-				name, shortName, tag, err = containers.SplitImageName(imageSpec)
+				name, registry, shortName, tag, err = containers.SplitImageName(imageSpec)
 				if err != nil {
 					log.Debugf("cannot split image name %q for container %q: %s", imageSpec, container.ID, err)
 					return image
@@ -315,6 +316,7 @@ func extractImage(ctx context.Context, container types.ContainerJSON, resolve re
 	}
 
 	image.Name = name
+	image.Registry = registry
 	image.ShortName = shortName
 	image.Tag = tag
 
diff --git a/pkg/workloadmeta/collectors/internal/podman/podman_test.go b/pkg/workloadmeta/collectors/internal/podman/podman_test.go
index 22b3256485a88..c0253533af0df 100644
--- a/pkg/workloadmeta/collectors/internal/podman/podman_test.go
+++ b/pkg/workloadmeta/collectors/internal/podman/podman_test.go
@@ -173,6 +173,7 @@ func TestPull(t *testing.T) {
 				Image: workloadmeta.ContainerImage{
 					RawName:   "docker.io/datadog/agent:latest",
 					Name:      "docker.io/datadog/agent",
+					Registry:  "docker.io",
 					ShortName: "agent",
 					Tag:       "latest",
 				},
@@ -222,6 +223,7 @@ func TestPull(t *testing.T) {
 				Image: workloadmeta.ContainerImage{
 					RawName:   "docker.io/datadog/agent-dev:latest",
 					Name:      "docker.io/datadog/agent-dev",
+					Registry:  "docker.io",
 					ShortName: "agent-dev",
 					Tag:       "latest",
 				},
diff --git a/pkg/workloadmeta/types.go b/pkg/workloadmeta/types.go
index 5253474b27071..ec41085ac3a03 100644
--- a/pkg/workloadmeta/types.go
+++ b/pkg/workloadmeta/types.go
@@ -12,7 +12,7 @@ import (
 	"time"
 
 	"github.com/mohae/deepcopy"
-	"github.com/opencontainers/image-spec/specs-go/v1"
+	v1 "github.com/opencontainers/image-spec/specs-go/v1"
 
 	"github.com/DataDog/datadog-agent/pkg/util/containers"
 )
@@ -270,6 +270,7 @@ type ContainerImage struct {
 	ID        string
 	RawName   string
 	Name      string
+	Registry  string
 	ShortName string
 	Tag       string
 }
@@ -281,7 +282,7 @@ func NewContainerImage(imageName string) (ContainerImage, error) {
 		Name:    imageName,
 	}
 
-	name, shortName, tag, err := containers.SplitImageName(imageName)
+	name, registry, shortName, tag, err := containers.SplitImageName(imageName)
 	if err != nil {
 		return image, err
 	}
@@ -291,6 +292,7 @@ func NewContainerImage(imageName string) (ContainerImage, error) {
 	}
 
 	image.Name = name
+	image.Registry = registry
 	image.ShortName = shortName
 	image.Tag = tag
 
@@ -623,6 +625,7 @@ var _ Entity = &ECSTask{}
 type ContainerImageMetadata struct {
 	EntityID
 	EntityMeta
+	Registry     string
 	ShortName    string
 	RepoTags     []string
 	RepoDigests  []string
diff --git a/releasenotes/notes/container_image-22cddfd6c194def9.yaml b/releasenotes/notes/container_image-22cddfd6c194def9.yaml
new file mode 100644
index 0000000000000..4eb38aeb49d2c
--- /dev/null
+++ b/releasenotes/notes/container_image-22cddfd6c194def9.yaml
@@ -0,0 +1,11 @@
+# Each section from every release note are combined when the
+# CHANGELOG.rst is rendered. So the text needs to be worded so that
+# it does not depend on any information only available in another
+# section. This may mean repeating some details, but each section
+# must be readable independently of the other.
+#
+# Each section note must be formatted as reStructuredText.
+---
+features:
+  - |
+    Add a new ``container_image`` long running check to collect information about container images.