Add support for port mapping in docker hints (#31243)

CHANGELOG.next.asciidoc

@@ -105,6 +105,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...main[Check the HEAD dif
 - Add action_input_type for the .fleet-actions-results {pull}30562[30562]
 - Add cronjob metadata by default {pull}30637[30637]
 - New option `setup.template.json.data_stream` is added to indicate if the JSON index template is a data stream. {pull}31048[31048]
+- Add support for port mapping in docker hints. {pull}31243[31243]
 
 *Auditbeat*
 

libbeat/autodiscover/providers/docker/docker.go

@@ -21,11 +21,12 @@
 package docker
 
 import (
+	"errors"
 	"fmt"
+	"strconv"
 	"time"
 
 	"github.com/gofrs/uuid"
-	"github.com/pkg/errors"
 
 	"github.com/elastic/beats/v7/libbeat/autodiscover"
 	"github.com/elastic/beats/v7/libbeat/autodiscover/builder"
@@ -39,7 +40,7 @@ import (
 )
 
 func init() {
-	autodiscover.Registry.AddProvider("docker", AutodiscoverBuilder)
+	_ = autodiscover.Registry.AddProvider("docker", AutodiscoverBuilder)
 }
 
 // Provider implements autodiscover provider for docker containers
@@ -70,7 +71,7 @@ func AutodiscoverBuilder(
 	logger := logp.NewLogger("docker")
 
 	errWrap := func(err error) error {
-		return errors.Wrap(err, "error setting up docker autodiscover provider")
+		return fmt.Errorf("error setting up docker autodiscover provider: %w", err)
 	}
 
 	config := defaultConfig()
@@ -175,20 +176,29 @@ type dockerMetadata struct {
 func (d *Provider) generateMetaDocker(event bus.Event) (*docker.Container, *dockerMetadata) {
 	container, ok := event["container"].(*docker.Container)
 	if !ok {
-		d.logger.Error(errors.New("Couldn't get a container from watcher event"))
+		d.logger.Error(errors.New("couldn't get a container from watcher event"))
 		return nil, nil
 	}
 
 	// Don't dedot selectors, dedot only metadata used for events enrichment
 	labelMap := common.MapStr{}
 	metaLabelMap := common.MapStr{}
 	for k, v := range container.Labels {
-		safemapstr.Put(labelMap, k, v)
+		err := safemapstr.Put(labelMap, k, v)
+		if err != nil {
+			d.logger.Debugf("error adding k:v (%v:%v): %v", k, v, err)
+		}
 		if d.config.Dedot {
 			label := common.DeDot(k)
-			metaLabelMap.Put(label, v)
+			_, err := metaLabelMap.Put(label, v)
+			if err != nil {
+				d.logger.Debugf("error adding value (%v): %v", v, err)
+			}
 		} else {
-			safemapstr.Put(metaLabelMap, k, v)
+			err := safemapstr.Put(metaLabelMap, k, v)
+			if err != nil {
+				d.logger.Debugf("error adding k:v (%v:%v): %v", k, v, err)
+			}
 		}
 	}
 
@@ -265,20 +275,19 @@ func (d *Provider) scheduleStopContainer(event bus.Event) {
 }
 
 func (d *Provider) stopContainer(container *docker.Container, meta *dockerMetadata) {
-	if _, ok := d.stoppers[container.ID]; ok {
-		delete(d.stoppers, container.ID)
-	}
+	delete(d.stoppers, container.ID)
 
 	d.emitContainer(container, meta, "stop")
 }
 
 func (d *Provider) emitContainer(container *docker.Container, meta *dockerMetadata, flag string) {
 	var host string
+	var ports common.MapStr
 	if len(container.IPAddresses) > 0 {
 		host = container.IPAddresses[0]
 	}
 
-	var events []bus.Event
+	events := make([]bus.Event, 0)
 	// Without this check there would be overlapping configurations with and without ports.
 	if len(container.Ports) == 0 {
 		event := bus.Event{
@@ -292,16 +301,22 @@ func (d *Provider) emitContainer(container *docker.Container, meta *dockerMetada
 		}
 
 		events = append(events, event)
+	} else {
+		ports = common.MapStr{}
+		for _, port := range container.Ports {
+			ports[strconv.FormatUint(uint64(port.PrivatePort), 10)] = port.PublicPort
+		}
 	}
-
 	// Emit container container and port information
+
 	for _, port := range container.Ports {
 		event := bus.Event{
 			"provider":  d.uuid,
 			"id":        container.ID,
 			flag:        true,
 			"host":      host,
 			"port":      port.PrivatePort,
+			"ports":     ports,
 			"docker":    meta.Docker,
 			"container": meta.Container,
 			"meta":      meta.Metadata,
@@ -334,6 +349,7 @@ func (d *Provider) publish(events []bus.Event) {
 	event := bus.Event(common.MapStr(events[0]).Clone())
 	// Remove the port to avoid ambiguity during debugging
 	delete(event, "port")
+	delete(event, "ports")
 	event["config"] = configs
 
 	// Call all appenders to append any extra configuration
@@ -346,10 +362,13 @@ func (d *Provider) generateHints(event bus.Event) bus.Event {
 	// Builders are Beat specific.
 	e := bus.Event{}
 	var dockerMeta common.MapStr
+	var ok bool
 
 	if rawDocker, err := common.MapStr(event).GetValue("docker.container"); err == nil {
-		dockerMeta = rawDocker.(common.MapStr)
-		e["container"] = dockerMeta
+		dockerMeta, ok = rawDocker.(common.MapStr)
+		if ok {
+			e["container"] = dockerMeta
+		}
 	}
 
 	if host, ok := event["host"]; ok {
@@ -358,6 +377,9 @@ func (d *Provider) generateHints(event bus.Event) bus.Event {
 	if port, ok := event["port"]; ok {
 		e["port"] = port
 	}
+	if ports, ok := event["ports"]; ok {
+		e["ports"] = ports
+	}
 	if labels, err := dockerMeta.GetValue("labels"); err == nil {
 		hints := builder.GenerateHints(labels.(common.MapStr), "", d.config.Prefix)
 		e["hints"] = hints

metricbeat/docs/autodiscover-hints.asciidoc

@@ -17,6 +17,9 @@ For Kuberentes autodiscover events users can leverage port names as well,
 like `http://${data.host}:${data.ports.prometheus}/metrics`.
 In the last one we refer to the container port with name `prometheus`.
 
+In order to target one specific exposed port `localhost:{data.ports.8222}` can be used
+in template config, where `8222` is the internal container port of the exposed targeted port.
+
 [float]
 ===== `co.elastic.metrics/metricsets`