internal/elasticsearch/ingest: export LoadIngestPipelineFiles and InstallPipelinesInElasticsearch

Author: efd6

internal/elasticsearch/ingest/datastream.go

@@ -58,19 +58,22 @@ func InstallDataStreamPipelines(ctx context.Context, api *elasticsearch.API, dat
 	nonce := time.Now().UnixNano()
 
 	mainPipeline := getPipelineNameWithNonce(dataStreamManifest.GetPipelineNameOrDefault(), nonce)
-	pipelines, err := loadIngestPipelineFiles(dataStreamPath, nonce)
+	pipelines, err := LoadIngestPipelineFiles(dataStreamPath, nonce)
 	if err != nil {
 		return "", nil, fmt.Errorf("loading ingest pipeline files failed: %w", err)
 	}
 
-	err = installPipelinesInElasticsearch(ctx, api, pipelines)
+	err = InstallPipelinesInElasticsearch(ctx, api, pipelines)
 	if err != nil {
 		return "", nil, err
 	}
 	return mainPipeline, pipelines, nil
 }
 
-func loadIngestPipelineFiles(dataStreamPath string, nonce int64) ([]Pipeline, error) {
+// LoadIngestPipelineFiles returns the set of pipelines found in the directory
+// elasticsearch/ingest_pipeline under the provided data stream path. The names
+// of the pipelines are decorated with the provided nonce.
+func LoadIngestPipelineFiles(dataStreamPath string, nonce int64) ([]Pipeline, error) {
 	elasticsearchPath := filepath.Join(dataStreamPath, "elasticsearch", "ingest_pipeline")
 
 	var pipelineFiles []string
@@ -241,7 +244,9 @@ func convertValue(value interface{}, label string) ([]string, error) {
 	}
 }
 
-func installPipelinesInElasticsearch(ctx context.Context, api *elasticsearch.API, pipelines []Pipeline) error {
+// InstallPipelinesInElasticsearch installs the provided pipelines into the
+// Elasticsearch instance specified by the provided API handle.
+func InstallPipelinesInElasticsearch(ctx context.Context, api *elasticsearch.API, pipelines []Pipeline) error {
 	for _, p := range pipelines {
 		if err := installPipeline(ctx, api, p); err != nil {
 			return err