fix pipeline names

apmpackage/apm/0.1.0/data_stream/internal_metrics/elasticsearch/ingest_pipeline/default.json

@@ -0,0 +1,25 @@
+{
+  "description": "Default enrichment for APM events",
+  "processors": [
+    {
+      "pipeline": {
+        "name": "metrics-apm.internal-0.1.0-apm_user_agent"
+      }
+    },
+    {
+      "pipeline": {
+        "name": "metrics-apm.internal-0.1.0-apm_user_geo"
+      }
+    },
+    {
+      "pipeline": {
+        "name": "metrics-apm.internal-0.1.0-apm_ingest_timestamp"
+      }
+    },
+    {
+      "pipeline": {
+        "name": "metrics-apm.internal-0.1.0-apm_remove_span_metadata"
+      }
+    }
+  ]
+}

apmpackage/apm/0.1.0/data_stream/internal_metrics/manifest.yml

@@ -1,4 +1,3 @@
 title: APM internal metrics
 type: metrics
 dataset: apm.internal
-ingest_pipeline: apm

apmpackage/apm/0.1.0/data_stream/logs/elasticsearch/ingest_pipeline/default.json

@@ -3,22 +3,22 @@
   "processors": [
     {
       "pipeline": {
-        "name": "apm_user_agent"
+        "name": "logs-apm.error-0.1.0-apm_user_agent"
       }
     },
     {
       "pipeline": {
-        "name": "apm_user_geo"
+        "name": "logs-apm.error-0.1.0-apm_user_geo"
       }
     },
     {
       "pipeline": {
-        "name": "apm_ingest_timestamp"
+        "name": "logs-apm.error-0.1.0-apm_ingest_timestamp"
       }
     },
     {
       "pipeline": {
-        "name": "apm_remove_span_metadata"
+        "name": "logs-apm.error-0.1.0-apm_remove_span_metadata"
       }
     }
   ]

apmpackage/apm/0.1.0/data_stream/metrics/elasticsearch/ingest_pipeline/apm.json → apmpackage/apm/0.1.0/data_stream/metrics/elasticsearch/ingest_pipeline/default.json

@@ -3,22 +3,22 @@
   "processors": [
     {
       "pipeline": {
-        "name": "apm_user_agent"
+        "name": "metrics-apm-0.1.0-apm_user_agent"
       }
     },
     {
       "pipeline": {
-        "name": "apm_user_geo"
+        "name": "metrics-apm-0.1.0-apm_user_geo"
       }
     },
     {
       "pipeline": {
-        "name": "apm_ingest_timestamp"
+        "name": "metrics-apm-0.1.0-apm_ingest_timestamp"
       }
     },
     {
       "pipeline": {
-        "name": "apm_remove_span_metadata"
+        "name": "metrics-apm-0.1.0-apm_remove_span_metadata"
       }
     }
   ]

apmpackage/apm/0.1.0/data_stream/metrics/manifest.yml

@@ -1,4 +1,3 @@
 title: APM application metrics
 type: metrics
 dataset: apm
-ingest_pipeline: apm

apmpackage/apm/0.1.0/data_stream/profiles/elasticsearch/ingest_pipeline/apm.json → apmpackage/apm/0.1.0/data_stream/profiles/elasticsearch/ingest_pipeline/default.json

@@ -3,22 +3,22 @@
   "processors": [
     {
       "pipeline": {
-        "name": "apm_user_agent"
+        "name": "profiles-apm-0.1.0-apm_user_agent"
       }
     },
     {
       "pipeline": {
-        "name": "apm_user_geo"
+        "name": "profiles-apm-0.1.0-apm_user_geo"
       }
     },
     {
       "pipeline": {
-        "name": "apm_ingest_timestamp"
+        "name": "profiles-apm-0.1.0-apm_ingest_timestamp"
       }
     },
     {
       "pipeline": {
-        "name": "apm_remove_span_metadata"
+        "name": "profiles-apm-0.1.0-apm_remove_span_metadata"
       }
     }
   ]

apmpackage/apm/0.1.0/data_stream/profiles/manifest.yml

@@ -1,4 +1,3 @@
 title: APM profiles
 type: metrics
 dataset: apm.profiling
-ingest_pipeline: apm

apmpackage/apm/0.1.0/data_stream/internal_metrics/elasticsearch/ingest_pipeline/apm.json → apmpackage/apm/0.1.0/data_stream/traces/elasticsearch/ingest_pipeline/default.json

@@ -3,22 +3,22 @@
   "processors": [
     {
       "pipeline": {
-        "name": "apm_user_agent"
+        "name": "traces-apm-0.1.0-apm_user_agent"
       }
     },
     {
       "pipeline": {
-        "name": "apm_user_geo"
+        "name": "traces-apm-0.1.0-apm_user_geo"
       }
     },
     {
       "pipeline": {
-        "name": "apm_ingest_timestamp"
+        "name": "traces-apm-0.1.0-apm_ingest_timestamp"
       }
     },
     {
       "pipeline": {
-        "name": "apm_remove_span_metadata"
+        "name": "traces-apm-0.1.0-apm_remove_span_metadata"
       }
     }
   ]

apmpackage/apm/0.1.0/data_stream/traces/manifest.yml

@@ -1,4 +1,3 @@
 title: APM traces
 type: traces
 dataset: apm
-ingest_pipeline: apm

apmpackage/cmd/gen-package/genpipelines.go

@@ -19,11 +19,68 @@ package main
 
 import (
 	"encoding/json"
+	"errors"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 )
 
+var streamMappings = map[string]string{
+	"logs":             "logs-apm.error",
+	"traces":           "traces-apm",
+	"metrics":          "metrics-apm",
+	"internal_metrics": "metrics-apm.internal",
+	"profiles":         "profiles-apm",
+}
+
+type PipelineDef struct {
+	ID   string       `json:"id"`
+	Body PipelineBody `json:"body"`
+}
+
+type PipelineBody struct {
+	Description string      `json:"description"`
+	Processors  []Processor `json:"processors"`
+}
+
+type Processor struct {
+	Pipeline *Pipeline `json:"pipeline,omitempty"`
+	m        map[string]interface{}
+}
+
+type Pipeline struct {
+	Name string `json:"name"`
+}
+
+type _Processor Processor
+
+func (p *Processor) UnmarshalJSON(bytes []byte) error {
+	aux := _Processor{}
+	err := json.Unmarshal(bytes, &aux)
+	if err != nil {
+		return err
+	}
+
+	*p = Processor(aux)
+	m := make(map[string]interface{})
+
+	err = json.Unmarshal(bytes, &m)
+	if err != nil {
+		return err
+	}
+	delete(m, "pipeline")
+	p.m = m
+	return nil
+}
+
+func (p *Processor) MarshalJSON() ([]byte, error) {
+	aux := _Processor(*p)
+	if p.Pipeline != nil {
+		return json.Marshal(aux)
+	}
+	return json.Marshal(p.m)
+}
+
 func generatePipelines(version, dataStream string) {
 	pipelines, err := os.Open("ingest/pipeline/definition.json")
 	if err != nil {
@@ -36,29 +93,40 @@ func generatePipelines(version, dataStream string) {
 		panic(err)
 	}
 
-	var definitions = make([]map[string]interface{}, 0)
+	var definitions = make([]PipelineDef, 0)
 	err = json.Unmarshal(bytes, &definitions)
 	if err != nil {
 		panic(err)
 	}
 
 	os.MkdirAll(pipelinesPath(version, dataStream), 0755)
 
+	var apmPipeline PipelineBody
 	for _, definition := range definitions {
-		pipeline, ok := definition["body"]
-		if !ok {
-			continue
-		}
-		id, ok := definition["id"]
-		if !ok {
+		pipeline := definition.Body
+		if definition.ID == "apm" {
+			apmPipeline = pipeline
 			continue
 		}
-
 		out, err := json.MarshalIndent(pipeline, "", "  ")
 		if err != nil {
 			panic(err)
 		}
-		fName := filepath.Join(pipelinesPath(version, dataStream), id.(string)+".json")
+		fName := filepath.Join(pipelinesPath(version, dataStream), definition.ID+".json")
 		ioutil.WriteFile(fName, out, 0644)
 	}
+
+	for _, p := range apmPipeline.Processors {
+		if p.Pipeline == nil {
+			// should not happen, lets panic loudly
+			panic(errors.New("expected pipeline processor"))
+		}
+		p.Pipeline.Name = streamMappings[dataStream] + "-" + version + "-" + p.Pipeline.Name
+	}
+	out, err := json.MarshalIndent(apmPipeline, "", "  ")
+	if err != nil {
+		panic(err)
+	}
+	fName := filepath.Join(pipelinesPath(version, dataStream), "default.json")
+	ioutil.WriteFile(fName, out, 0644)
 }

apmpackage/cmd/gen-package/main.go

@@ -23,7 +23,6 @@ import (
 	"io/ioutil"
 	"log"
 	"os"
-	"path/filepath"
 
 	"github.com/elastic/apm-server/cmd"
 	"github.com/elastic/beats/v7/libbeat/common"
@@ -46,11 +45,6 @@ func main() {
 	for dataStream := range inputFields {
 		generatePipelines(packageVersion, dataStream)
 	}
-	// hack, remove when bugfix comes to Kibana
-	bad := filepath.Join(pipelinesPath(packageVersion, "logs"), "apm.json")
-	good := filepath.Join(pipelinesPath(packageVersion, "logs"), "default.json")
-	os.Rename(bad, good)
-
 	generateDocs(inputFields, packageVersion)
 	log.Printf("Package fields and docs generated for version %s (stack %s)", packageVersion, stackVersion.String())
 }
@@ -66,6 +60,7 @@ func clear(version string) {
 		if f.IsDir() {
 			os.Remove(ecsFilePath(version, f.Name()))
 			os.Remove(fieldsFilePath(version, f.Name()))
+			os.RemoveAll(pipelinesPath(version, f.Name()))
 		}
 	}
 	ioutil.WriteFile(docsFilePath(version), nil, 0644)