Add regex based cid extractor to add_process_metadata

CHANGELOG.next.asciidoc

@@ -167,6 +167,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Update RPM packages contained in Beat Docker images. {issue}17035[17035]
 - Update supported versions of `redis` output. {pull}17198[17198]
 - Update documentation for system.process.memory fields to include clarification on Windows os's. {pull}17268[17268]
+- Add optional regex based cid extractor to `add_kubernetes_metadata` processor. {pull}17360[17360]
 
 *Auditbeat*
 

libbeat/processors/add_process_metadata/add_process_metadata.go

@@ -126,9 +126,9 @@ func newProcessMetadataProcessorWithProvider(cfg *common.Config, provider proces
 
 			cgroupsCache := common.NewCacheWithRemovalListener(config.CgroupCacheExpireTime, 100, evictionListener)
 			cgroupsCache.StartJanitor(config.CgroupCacheExpireTime)
-			p.cidProvider = newCidProvider(config.HostPath, config.CgroupPrefixes, processCgroupPaths, cgroupsCache)
+			p.cidProvider = newCidProvider(config.HostPath, config.CgroupPrefixes, config.CgroupRegex, processCgroupPaths, cgroupsCache)
 		} else {
-			p.cidProvider = newCidProvider(config.HostPath, config.CgroupPrefixes, processCgroupPaths, nil)
+			p.cidProvider = newCidProvider(config.HostPath, config.CgroupPrefixes, config.CgroupRegex, processCgroupPaths, nil)
 		}
 
 	}

libbeat/processors/add_process_metadata/add_process_metadata_test.go

@@ -485,6 +485,31 @@ func TestAddProcessMetadata(t *testing.T) {
 				},
 			},
 		},
+		{
+			description: "container.id based on regex in config",
+			config: common.MapStr{
+				"match_pids":     []string{"system.process.ppid"},
+				"include_fields": []string{"container.id"},
+				"cgroup_regex":   "\\/.+\\/.+\\/.+\\/([0-9a-f]{64}).*",
+			},
+			event: common.MapStr{
+				"system": common.MapStr{
+					"process": common.MapStr{
+						"ppid": "1",
+					},
+				},
+			},
+			expected: common.MapStr{
+				"system": common.MapStr{
+					"process": common.MapStr{
+						"ppid": "1",
+					},
+				},
+				"container": common.MapStr{
+					"id": "b5285682fba7449c86452b89a800609440ecc88a7ba5f2d38bedfb85409b30b1",
+				},
+			},
+		},
 		{
 			description: "without cgroup cache",
 			config: common.MapStr{

libbeat/processors/add_process_metadata/config.go

@@ -51,6 +51,9 @@ type config struct {
 	// CgroupPrefix is the prefix where the container id is inside cgroup
 	CgroupPrefixes []string `config:"cgroup_prefixes"`
 
+	// CgroupRegex is the regular expression that captures the container id from cgroup path
+	CgroupRegex string `config:"cgroup_regex"`
+
 	// CgroupCacheExpireTime is the length of time before cgroup cache elements expire in seconds,
 	// set to 0 to disable the cgroup cache
 	CgroupCacheExpireTime time.Duration `config:"cgroup_cache_expire_time"`

libbeat/processors/add_process_metadata/docs/add_process_metadata.asciidoc

@@ -80,6 +80,11 @@ to `/kubepods` and `/docker`. This is the prefix where the container ID is
 inside cgroup. For different runtime configurations of Kubernetes or Docker,
 the `cgroup_prefixes` can be set to overwrite the defaults.
 
+`cgroup_regex`:: (Optional) By default, the container id is extracted from 
+cgroup file based on `cgroup_prefixes`. This can be overwritten by specifying 
+regular expression with capture group for capturing container id from cgroup 
+path. For example: `^\/.+\/.+\/.+\/([0-9a-f]{64}).*`
+
 `cgroup_cache_expire_time`:: (Optional) By default, the 
 `cgroup_cache_expire_time` is set to 30 seconds. This is the length of time
 before cgroup cache elements expire in seconds. It can be set to 0 to disable

libbeat/processors/add_process_metadata/gosigar_cid_provider.go

@@ -20,6 +20,7 @@ package add_process_metadata
 import (
 	"os"
 	"path/filepath"
+	"regexp"
 	"strings"
 
 	"github.com/pkg/errors"
@@ -36,6 +37,7 @@ type gosigarCidProvider struct {
 	log                *logp.Logger
 	hostPath           string
 	cgroupPrefixes     []string
+	cgroupRegex        string
 	processCgroupPaths func(string, int) (map[string]string, error)
 	pidCidCache        *common.Cache
 }
@@ -68,11 +70,12 @@ func (p gosigarCidProvider) GetCid(pid int) (result string, err error) {
 	return cid, nil
 }
 
-func newCidProvider(hostPath string, cgroupPrefixes []string, processCgroupPaths func(string, int) (map[string]string, error), pidCidCache *common.Cache) gosigarCidProvider {
+func newCidProvider(hostPath string, cgroupPrefixes []string, cgroupRegex string, processCgroupPaths func(string, int) (map[string]string, error), pidCidCache *common.Cache) gosigarCidProvider {
 	return gosigarCidProvider{
 		log:                logp.NewLogger(providerName),
 		hostPath:           hostPath,
 		cgroupPrefixes:     cgroupPrefixes,
+		cgroupRegex:        cgroupRegex,
 		processCgroupPaths: processCgroupPaths,
 		pidCidCache:        pidCidCache,
 	}
@@ -104,12 +107,25 @@ func (p gosigarCidProvider) getProcessCgroups(pid int) (map[string]string, error
 // Example:
 // /kubepods/besteffort/pod9b9e44c2-00fd-11ea-95e9-080027421ddf/2bb9fd4de339e5d4f094e78bb87636004acfe53f5668104addc761fe4a93588e
 func (p gosigarCidProvider) getCid(cgroups map[string]string) string {
-	for _, path := range cgroups {
-		for _, prefix := range p.cgroupPrefixes {
-			if strings.HasPrefix(path, prefix) {
-				return filepath.Base(path)
+	// if regex defined use it to find cid
+	if len(p.cgroupRegex) != 0 {
+		re := regexp.MustCompile(p.cgroupRegex)
+		for _, path := range cgroups {
+			rs := re.FindStringSubmatch(path)
+			if rs != nil {
+				return rs[1]
 			}
 		}
+	} else {
+		// use string prefix to find cid
+		for _, path := range cgroups {
+			for _, prefix := range p.cgroupPrefixes {
+				if strings.HasPrefix(path, prefix) {
+					return filepath.Base(path)
+				}
+			}
+
+		}
 	}
 	return ""
 }