Adds reprovision API to support updating search pipelines, ingest pipelines index settings

[joshpalis]

Description

Adds fine-grained provisioning API (reprovision) to allow users to iteratively update already provisioned templates without having to first deprovision their created resources. The diagram below illustrates the execution flow of a reprovision request.

URI

• PUT /_plugins/_flow_framework/workflow/<workflow_id>?reprovision=true

Capabilities

• Allows a user to update existing template resources created through Flow Framework with the reprovision=true parameter
• Allows users to add new resources to existing templates
• Allows users to add new predecessor workflow nodes to existing workflow nodes
• Does not allow deleting nodes (fine-grained deprovisioning)

Supported Update Steps

• Update Index Settings
• Update Search Pipeline
• Update Ingest Pipeline

Note : Attempts to update non-supported steps will result in an error message

Examples :

Iteratively adding new nodes and updating nodes

1. Register local model only

curl -i -XPOST "localhost:9200/_plugins/_flow_framework/workflow?provision=true" -H "Content-Type:application/json" --data '{"name":"semantic search with local pretrained model","description":"Setting up semantic search, with a local pretrained embedding model","use_case":"SEMANTIC_SEARCH","version":{"template":"1.0.0","compatibility":["2.12.0","3.0.0"]},"workflows":{"provision":{"nodes":[{"id":"register_local_pretrained_model","type":"register_local_pretrained_model","user_inputs":{"name":"huggingface/sentence-transformers/paraphrase-MiniLM-L3-v2","version":"1.0.1","description":"This is a sentence transformer model","model_format":"TORCH_SCRIPT","deploy":true}}]}}}'
HTTP/1.1 201 Created
X-OpenSearch-Version: OpenSearch/3.0.0-SNAPSHOT (opensearch)
content-type: application/json; charset=UTF-8
content-length: 38

{"workflow_id":"buI-65ABYsYquLNfDj2t"}

2. Add text embedding ingest pipeline which uses the model

curl -i -XPUT "localhost:9200/_plugins/_flow_framework/workflow/buI-65ABYsYquLNfDj2t?reprovision=true" -H "Content-Type:application/json" --data '{"name":"semantic search with local pretrained model","description":"Setting up semantic search, with a local pretrained embedding model","use_case":"SEMANTIC_SEARCH","version":{"template":"1.0.0","compatibility":["2.12.0","3.0.0"]},"workflows":{"provision":{"nodes":[{"id":"register_local_pretrained_model","type":"register_local_pretrained_model","user_inputs":{"name":"huggingface/sentence-transformers/paraphrase-MiniLM-L3-v2","version":"1.0.1","description":"This is a sentence transformer model","model_format":"TORCH_SCRIPT","deploy":true}},{"id":"create_ingest_pipeline","type":"create_ingest_pipeline","previous_node_inputs":{"register_local_pretrained_model":"model_id"},"user_inputs":{"pipeline_id":"nlp-ingest-pipeline","configurations":{"description":"A text embedding pipeline","processors":[{"text_embedding":{"model_id":"${{register_local_pretrained_model.model_id}}","field_map":{"passage_text":"passage_embedding"}}}]}}}]}}}'
HTTP/1.1 201 Created
X-OpenSearch-Version: OpenSearch/3.0.0-SNAPSHOT (opensearch)
content-type: application/json; charset=UTF-8
content-length: 38

{"workflow_id":"buI-65ABYsYquLNfDj2t"}

3. Create index which attaches the ingest pipeline as the default_pipeline

curl -i -XPUT "localhost:9200/_plugins/_flow_framework/workflow/buI-65ABYsYquLNfDj2t?reprovision=true" -H "Content-Type:application/json" --data '{"name":"semantic search with local pretrained model","description":"Setting up semantic search, with a local pretrained embedding model","use_case":"SEMANTIC_SEARCH","version":{"template":"1.0.0","compatibility":["2.12.0","3.0.0"]},"workflows":{"provision":{"nodes":[{"id":"register_local_pretrained_model","type":"register_local_pretrained_model","user_inputs":{"name":"huggingface/sentence-transformers/paraphrase-MiniLM-L3-v2","version":"1.0.1","description":"This is a sentence transformer model","model_format":"TORCH_SCRIPT","deploy":true}},{"id":"create_ingest_pipeline","type":"create_ingest_pipeline","previous_node_inputs":{"register_local_pretrained_model":"model_id"},"user_inputs":{"pipeline_id":"nlp-ingest-pipeline","configurations":{"description":"A text embedding pipeline","processors":[{"text_embedding":{"model_id":"${{register_local_pretrained_model.model_id}}","field_map":{"passage_text":"passage_embedding"}}}]}}},{"id":"create_index","type":"create_index","previous_node_inputs":{"create_ingest_pipeline":"pipeline_id"},"user_inputs":{"index_name":"my-nlp-index","configurations":{"settings":{"index.knn":true,"default_pipeline":"${{create_ingest_pipeline.pipeline_id}}","index.number_of_shards":"2"},"mappings":{"properties":{"passage_embedding":{"type":"knn_vector","dimension":"768","method":{"engine":"lucene","space_type":"l2","name":"hnsw","parameters":{}}},"passage_text":{"type":"text"}}}}}}]}}}'

HTTP/1.1 201 Created
X-OpenSearch-Version: OpenSearch/3.0.0-SNAPSHOT (opensearch)
content-type: application/json; charset=UTF-8
content-length: 38

{"workflow_id":"buI-65ABYsYquLNfDj2t"}

The workflow state is updated for each new resource added :

curl -i -XGET "localhost:9200/_plugins/_flow_framework/workflow/buI-65ABYsYquLNfDj2t/_status?pretty&&all=true"
HTTP/1.1 200 OK
X-OpenSearch-Version: OpenSearch/3.0.0-SNAPSHOT (opensearch)
content-type: application/json; charset=UTF-8
content-length: 1019

{
  "workflow_id" : "buI-65ABYsYquLNfDj2t",
  "state" : "COMPLETED",
  "provisioning_progress" : "DONE",
  "provision_start_time" : 1721933687578,
  "provision_end_time" : 1721933687634,
  "resources_created" : [
    {
      "resource_type" : "model_id",
      "resource_id" : "ceI-65ABYsYquLNfFD13",
      "workflow_step_name" : "register_local_pretrained_model",
      "workflow_step_id" : "register_local_pretrained_model"
    },
    {
      "resource_type" : "model_id",
      "resource_id" : "ceI-65ABYsYquLNfFD13",
      "workflow_step_name" : "deploy_model",
      "workflow_step_id" : "register_local_pretrained_model"
    },
    {
      "resource_type" : "pipeline_id",
      "resource_id" : "nlp-ingest-pipeline",
      "workflow_step_name" : "create_ingest_pipeline",
      "workflow_step_id" : "create_ingest_pipeline"
    },
    {
      "resource_type" : "index_name",
      "resource_id" : "my-nlp-index",
      "workflow_step_name" : "create_index",
      "workflow_step_id" : "create_index"
    }
  ]
}

4. Update the index settings to remove the default_pipeline. Index settings can be flattened or expanded

curl -i -XPUT "localhost:9200/_plugins/_flow_framework/workflow/buI-65ABYsYquLNfDj2t?reprovision=true" -H "Content-Type:application/json" --data '{"name":"semantic search with local pretrained model","description":"Setting up semantic search, with a local pretrained embedding model","use_case":"SEMANTIC_SEARCH","version":{"template":"1.0.0","compatibility":["2.12.0","3.0.0"]},"workflows":{"provision":{"nodes":[{"id":"register_local_pretrained_model","type":"register_local_pretrained_model","user_inputs":{"name":"huggingface/sentence-transformers/paraphrase-MiniLM-L3-v2","version":"1.0.1","description":"This is a sentence transformer model","model_format":"TORCH_SCRIPT","deploy":true}},{"id":"create_ingest_pipeline","type":"create_ingest_pipeline","previous_node_inputs":{"register_local_pretrained_model":"model_id"},"user_inputs":{"pipeline_id":"nlp-ingest-pipeline","configurations":{"description":"A text embedding pipeline","processors":[{"text_embedding":{"model_id":"${{register_local_pretrained_model.model_id}}","field_map":{"passage_text":"passage_embedding"}}}]}}},{"id":"create_index","type":"create_index","previous_node_inputs":{"create_ingest_pipeline":"pipeline_id"},"user_inputs":{"index_name":"my-nlp-index","configurations":{"settings":{"index.knn":true,"default_pipeline":"_none","index.number_of_shards":"2"},"mappings":{"properties":{"passage_embedding":{"type":"knn_vector","dimension":"768","method":{"engine":"lucene","space_type":"l2","name":"hnsw","parameters":{}}},"passage_text":{"type":"text"}}}}}}]}}}'

HTTP/1.1 201 Created
X-OpenSearch-Version: OpenSearch/3.0.0-SNAPSHOT (opensearch)
content-type: application/json; charset=UTF-8
content-length: 38

{"workflow_id":"buI-65ABYsYquLNfDj2t"}

Retrieving the index shows the change has been applied to the index settings :

curl -XGET "localhost:9200/my-nlp-index/_settings?pretty"
{
  "my-nlp-index" : {
    "settings" : {
      "index" : {
        "replication" : {
          "type" : "DOCUMENT"
        },
        "number_of_shards" : "2",
        "provided_name" : "my-nlp-index",
        "default_pipeline" : "_none",
        "knn" : "true",
        "creation_date" : "1721934815493",
        "number_of_replicas" : "1",
        "uuid" : "KqPrlkXPTNO0rcQg-vxMcA",
        "version" : {
          "created" : "137217827"
        }
      }
    }
  }
}

Adding new predecessor nodes to existing nodes (Mid-node addition)

1. Registers a model and creates an index (no edges)

curl -i -XPOST "localhost:9200/_plugins/_flow_framework/workflow?provision=true" -H "Content-Type:application/json" --data '{"name":"semantic search with local pretrained model","description":"Setting up semantic search, with a local pretrained embedding model","use_case":"SEMANTIC_SEARCH","version":{"template":"1.0.0","compatibility":["2.12.0","3.0.0"]},"workflows":{"provision":{"nodes":[{"id":"register_local_pretrained_model","type":"register_local_pretrained_model","user_inputs":{"name":"huggingface/sentence-transformers/paraphrase-MiniLM-L3-v2","version":"1.0.1","description":"This is a sentence transformer model","model_format":"TORCH_SCRIPT","deploy":true}},{"id":"create_index","type":"create_index","user_inputs":{"index_name":"my-nlp-index","configurations":{"settings":{"index.knn":true,"index.number_of_shards":"2"},"mappings":{"properties":{"passage_embedding":{"type":"knn_vector","dimension":"768","method":{"engine":"lucene","space_type":"l2","name":"hnsw","parameters":{}}},"passage_text":{"type":"text"}}}}}}]}}}'
HTTP/1.1 201 Created
X-OpenSearch-Version: OpenSearch/3.0.0-SNAPSHOT (opensearch)
content-type: application/json; charset=UTF-8
content-length: 38

{"workflow_id":"R0ZS65AB_HLxu0qR5nJJ"}

2. Adds an ingest pipeline which uses the model and sets default pipeline to index

curl -i -XPUT "localhost:9200/_plugins/_flow_framework/workflow/R0ZS65AB_HLxu0qR5nJJ?reprovision=true" -H "Content-Type:application/json" --data '{"name":"semantic search with local pretrained model","description":"Setting up semantic search, with a local pretrained embedding model","use_case":"SEMANTIC_SEARCH","version":{"template":"1.0.0","compatibility":["2.12.0","3.0.0"]},"workflows":{"provision":{"nodes":[{"id":"register_local_pretrained_model","type":"register_local_pretrained_model","user_inputs":{"name":"huggingface/sentence-transformers/paraphrase-MiniLM-L3-v2","version":"1.0.1","description":"This is a sentence transformer model","model_format":"TORCH_SCRIPT","deploy":true}},{"id":"create_ingest_pipeline","type":"create_ingest_pipeline","previous_node_inputs":{"register_local_pretrained_model":"model_id"},"user_inputs":{"pipeline_id":"nlp-ingest-pipeline","configurations":{"description":"A text embedding pipeline","processors":[{"text_embedding":{"model_id":"${{register_local_pretrained_model.model_id}}","field_map":{"passage_text":"passage_embedding"}}}]}}},{"id":"create_index","type":"create_index","previous_node_inputs":{"create_ingest_pipeline":"pipeline_id"},"user_inputs":{"index_name":"my-nlp-index","configurations":{"settings":{"index.knn":true,"index.default_pipeline":"${{create_ingest_pipeline.pipeline_id}}","index.number_of_shards":"2"},"mappings":{"properties":{"passage_embedding":{"type":"knn_vector","dimension":"768","method":{"engine":"lucene","space_type":"l2","name":"hnsw","parameters":{}}},"passage_text":{"type":"text"}}}}}}]}}}'

HTTP/1.1 201 Created
X-OpenSearch-Version: OpenSearch/3.0.0-SNAPSHOT (opensearch)
content-type: application/json; charset=UTF-8
content-length: 38

{"workflow_id":"R0ZS65AB_HLxu0qR5nJJ"}

Default pipeline is attached to index :

 curl -XGET "localhost:9200/my-nlp-index/_settings?pretty"
{
  "my-nlp-index" : {
    "settings" : {
      "index" : {
        "replication" : {
          "type" : "DOCUMENT"
        },
        "number_of_shards" : "2",
        "provided_name" : "my-nlp-index",
        "default_pipeline" : "nlp-ingest-pipeline",
        "knn" : "true",
        "creation_date" : "1721934997273",
        "number_of_replicas" : "1",
        "uuid" : "38oECrlETECF6AEcVZcCGg",
        "version" : {
          "created" : "137217827"
        }
      }
    }
  }
}

Resources created :

curl -i -XGET "localhost:9200/_plugins/_flow_framework/workflow/R0ZS65AB_HLxu0qR5nJJ/_status?pretty&&all=true"
HTTP/1.1 200 OK
X-OpenSearch-Version: OpenSearch/3.0.0-SNAPSHOT (opensearch)
content-type: application/json; charset=UTF-8
content-length: 1019

{
  "workflow_id" : "R0ZS65AB_HLxu0qR5nJJ",
  "state" : "COMPLETED",
  "provisioning_progress" : "DONE",
  "provision_start_time" : 1721935022741,
  "provision_end_time" : 1721935022863,
  "resources_created" : [
    {
      "resource_type" : "index_name",
      "resource_id" : "my-nlp-index",
      "workflow_step_name" : "create_index",
      "workflow_step_id" : "create_index"
    },
    {
      "resource_type" : "model_id",
      "resource_id" : "SkZS65AB_HLxu0qR6nJt",
      "workflow_step_name" : "register_local_pretrained_model",
      "workflow_step_id" : "register_local_pretrained_model"
    },
    {
      "resource_type" : "model_id",
      "resource_id" : "SkZS65AB_HLxu0qR6nJt",
      "workflow_step_name" : "deploy_model",
      "workflow_step_id" : "register_local_pretrained_model"
    },
    {
      "resource_type" : "pipeline_id",
      "resource_id" : "nlp-ingest-pipeline",
      "workflow_step_name" : "create_ingest_pipeline",
      "workflow_step_id" : "create_ingest_pipeline"
    }
  ]
}

Related Issues

Part of #717

Check List

• New functionality includes testing.
• New functionality has been documented.
• API changes companion pull request created.
• Commits are signed per the DCO using --signoff.
• Public documentation issue/PR created.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[codecov]

Codecov Report

Attention: Patch coverage is 72.68817% with 127 lines in your changes missing coverage. Please review.

Project coverage is 75.39%. Comparing base (7ec848a) to head (fed530b).

Files	Patch %	Lines
.../transport/ReprovisionWorkflowTransportAction.java	46.21%	70 Missing and 1 partial ⚠️
...search/flowframework/workflow/UpdateIndexStep.java	73.01%	16 Missing and 1 partial ⚠️
...ework/transport/CreateWorkflowTransportAction.java	73.91%	10 Missing and 2 partials ⚠️
.../flowframework/workflow/WorkflowProcessSorter.java	83.07%	6 Missing and 5 partials ⚠️
...search/flowframework/common/WorkflowResources.java	82.75%	2 Missing and 3 partials ⚠️
...earch/flowframework/transport/WorkflowRequest.java	68.75%	2 Missing and 3 partials ⚠️
...h/flowframework/rest/RestCreateWorkflowAction.java	69.23%	2 Missing and 2 partials ⚠️
...ramework/transport/ReprovisionWorkflowRequest.java	94.73%	1 Missing ⚠️
...framework/workflow/AbstractUpdatePipelineStep.java	96.66%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main     #804      +/-   ##
============================================
- Coverage     75.82%   75.39%   -0.44%     
- Complexity      838      917      +79     
============================================
  Files            88       96       +8     
  Lines          4054     4454     +400     
  Branches        373      410      +37     
============================================
+ Hits           3074     3358     +284     
- Misses          824      925     +101     
- Partials        156      171      +15     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[dbwiddis]

Mostly LGTM with a few nits. Some blockers:

• WorkflowRequest serialization breaks BWC
• Possible race condition with resetting error field

[amitgalitz]

One initial question, we are calling this a reprovision API but we are adding a new query param to the create/update API. Should we just have a new api that is a reprovision API. The provision query param we have added to the create API was to combine both APIs into one but here we are juts adding a query param to the update API.
I can also guess we can call this an improvement to allow update while a workflow is provisioned, just not sure if param represents this.

[amitgalitz]

On this example you give you say you are updating but you made a create_index step:

4. Update the index settings to remove the default_pipeline. Index settings can be flattened or expanded

Was this a mistake or we are creating the index again when updating?

[owaiskazi19]

Will continue the review

[joshpalis]

On this example you give you say you are updating but you made a create_index step:

> Update the index settings to remove the default_pipeline. Index settings can be flattened or expanded

Was this a mistake or we are creating the index again when updating?

Great question, it is intentional.

The user will always only be submitting a template with create steps to preserve idempotency. This way a single template can be used in any cluster to produce the same configuration. In the case of reprovisioning, we determine which create nodes have been modified and configure and execute the related update step (if any exists), somewhat obfuscating the fact that we're updating resources behind the scenes

[owaiskazi19]

Great work in implementing such design @joshpalis .

[dbwiddis]

LGTM with one small change needed.

[owaiskazi19]

LGTM