Op asset specificity

docs/jobs/configuration.md

@@ -40,7 +40,7 @@ The first operation in the [operations](#operations) list, reads from a particul
 
 | Configuration | Description                                                                                                                                                                                                              | Type     | Notes    |
 | ------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -------- | -------- |
-| `_name`       | The `_name` property is required, and it is required to be unqiue but can be suffixed with a identifier by using the format `"example:0"`, anything after the `:` is stripped out when searching for the file or folder. | `String` | required |
+| `_name`       | The `_name` property is required, and must match the name of api inside the asset. You may create several instances of an api and differentiate them by using the format `"example:someTag"`, and use the config `api_name: "example:someTag"` on the operation to reference the different apis. | `String` | required |
 
 ## Examples
 
@@ -98,4 +98,159 @@ The first operation in the [operations](#operations) list, reads from a particul
     ]
 }
 ```
+<!--Example with multiple apis->
+
+```json
+{
+    "name": "test",
+    "lifecycle": "once",
+    "workers": 1,
+    "analytics": true,
+    "assets": [
+        "elasticsearch:4.0.2"
+    ],
+    "apis": [
+        {
+            "_name": "elasticsearch_reader_api:foo",
+            "index": "ts_test_example-1000",
+            "size": 10000,
+            "date_field_name": "created",
+            "preserve_id": true
+        },
+        {
+            "_name": "elasticsearch_reader_api:bar",
+            "index": "some_other_index",
+            "size": 20000,
+            "date_field_name": "_ingest",
+            "preserve_id": false
+        }
+    ],
+    "operations": [
+        {
+            "_op": "elasticsearch_reader",
+            "api_name": "elasticsearch_reader_api:foo",
+            "index": "ts_test_example-1000",
+            "size": 10000,
+            "date_field_name": "created",
+            "preserve_id": true
+        },
+        {
+            "_op": "example_op",
+            "api_name": "elasticsearch_reader_api:bar"
+        }
+    ]
+}
+```
+<!--END_DOCUSAURUS_CODE_TABS-->
+
+### Operation and api name collisions
+When creating a job that uses multiple assets, we check to see if any of the apis or operations specified on the job are found in multiple assets. If they are found across multiple assets then we will throw an error on job submission as we cannot tell which asset to use.
+
+To determine which asset to use you must use the `@` symbol along with a valid asset identifier on the job to determine which asset to use when loading the operation or api. The asset identifier must match your name in the asset configuration on the job
+
+## Examples
+
+### Naming conventions
+
+If for example you have an asset listed as `standard` inside a job, the operation must be annotated with it. So for this asset and operation named `filter`, below are examples of valid names:
+
+- `filter@standard`
+
+If you list the asset as `standard:3.2.0`
+
+- `filter@standard:3.2.0`
+
+If you list the asset as its asset hash `2ab55a02723c304b2b74a7819942b4920e4ee6a9`
+
+- `filter@2ab55a02723c304b2b74a7819942b4920e4ee6a9`
+
+
+You can use multiple of the same asset name in a job as long as the operation and api names match which asset version you are trying to use. An example of this can be found in the example jobs below with two elasticsearch asset example.
+
+api names follow the same convention as operations with the only difference that apis allow for additional tags as mentioned [here](#apis).
+
+- `some_api:someAsset:1.1.0:foo`
+
+
+### Jobs
+In this example job, we are using two assets that both have an operation with the same `_op` named `filter` to be exact. This is an example job that shows how to use the correct naming conventions to determine which asset to use.
+
+<!--DOCUSAURUS_CODE_TABS-->
+<!-- job with an operation name that is shared between the two assets -->
+``` json
+{
+    "name": "test filter",
+    "lifecycle": "once",
+    "workers": 1,
+    "analytics": true,
+    "assets": [
+        "common_processors:0.16.0",
+        "standard:1.2.0"
+    ],
+    "operations": [
+        {
+            "_op": "data_generator",
+            "size": 1
+        },
+        {
+            "_op": "filter@common_processors:0.16.0",
+            "field": "ip",
+            "value": "0.0.0.0"
+        },
+        {
+            "_op": "noop"
+        }
+    ]
+}
+```
+<!--END_DOCUSAURUS_CODE_TABS-->
+
+In this example job, we are using two assets that both have an operation with the same `_op` and `api` name. This is an example job that show how to determine which asset to use for each api and operation.
+
+<!--DOCUSAURUS_CODE_TABS-->
+<!--Elasticsearch job with two different versions sharing both the operation and api -->
+``` json
+{
+    "name": "test",
+    "lifecycle": "once",
+    "workers": 1,
+    "analytics": true,
+    "assets": [
+        "elasticsearch:4.0.2",
+        "elasticsearch:4.0.5"
+    ],
+    "apis": [
+        {
+            "_name": "elasticsearch_sender_api@elasticsearch:4.0.5",
+            "index": "op_asset_version_test",
+            "preserve_id": true,
+            "size": 10000
+        },
+        {
+            "_name": "elasticsearch_reader_api@elasticsearch:4.0.2",
+            "index": "ts_test_example-1000",
+            "size": 10000,
+            "date_field_name": "created",
+            "preserve_id": true
+        }
+    ],
+    "operations": [
+        {
+            "_op": "elasticsearch_reader@elasticsearch:4.0.2",
+            "api_name": "elasticsearch_reader_api@elasticsearch:4.0.2",
+            "index": "ts_test_example-1000",
+            "size": 10000,
+            "date_field_name": "created",
+            "preserve_id": true
+        },
+        {
+            "_op": "elasticsearch_bulk@elasticsearch:4.0.5",
+            "api_name": "elasticsearch_sender_api@elasticsearch:4.0.5",
+            "index": "op_asset_version_test",
+            "preserve_id": true,
+            "size": 10000
+        }
+    ]
+}
+```
 <!--END_DOCUSAURUS_CODE_TABS-->

e2e/package.json

@@ -43,7 +43,7 @@
         "ms": "~2.1.3"
     },
     "devDependencies": {
-        "@terascope/types": "~1.3.1",
+        "@terascope/types": "~1.3.2",
         "bunyan": "~1.8.15",
         "elasticsearch-store": "~1.6.0",
         "fs-extra": "~11.2.0",

e2e/test/cases/assets/simple-spec.ts

@@ -1,7 +1,7 @@
 import 'jest-extended';
 import fs from 'node:fs';
-import os from 'os';
-import path from 'path';
+import os from 'node:os';
+import path from 'node:path';
 import decompress from 'decompress';
 import archiver from 'archiver';
 import {
@@ -72,6 +72,24 @@ describe('assets', () => {
             { blocking: true }
         );
 
+        /*
+            {
+                "name": "ex1",
+                "version": "0.0.1",
+                "node_version": 18,
+                "platform": false,
+                "arch": false
+            }
+
+            {
+                "name": "ex1",
+                "version": "0.1.1",
+                "node_version": 18,
+                "platform": false,
+                "arch": false
+            }
+
+        */
         const newerAssetPath = 'test/fixtures/assets/example_asset_1updated.zip';
         const fileStream = fs.createReadStream(newerAssetPath);
         // the asset on this job already points to 'ex1' so it should use the latest available asset
@@ -134,6 +152,57 @@ describe('assets', () => {
 
         await ex.stop({ blocking: true });
     });
+
+    it('will throw if there are naming conflicts', async () => {
+        const jobSpec = terasliceHarness.newJob('generator-asset');
+        // Set resource constraints on workers within CI
+        if (TEST_PLATFORM === 'kubernetes' || TEST_PLATFORM === 'kubernetesV2') {
+            jobSpec.resources_requests_cpu = 0.1;
+        }
+        // the previous test confirms the newer version will be used by default
+        // now we test to see if we can select the older version
+        jobSpec.assets = ['ex1:0.0.1', 'ex1:0.1.1', 'standard', 'elasticsearch'];
+
+        await expect(terasliceHarness.submitAndStart(jobSpec)).rejects.toThrow();
+    });
+
+    it('will not throw if there are naming conflicts but you use asset identifiers', async () => {
+        const jobSpec = terasliceHarness.newJob('generator-asset');
+        // Set resource constraints on workers within CI
+        if (TEST_PLATFORM === 'kubernetes' || TEST_PLATFORM === 'kubernetesV2') {
+            jobSpec.resources_requests_cpu = 0.1;
+        }
+        // the previous test confirms the newer version will be used by default
+        // now we test to see if we can select the older version
+        jobSpec.assets = ['ex1:0.0.1', 'ex1:0.1.1', 'standard', 'elasticsearch'];
+        jobSpec.operations = jobSpec.operations.map((op) => {
+            if (op._op === 'drop_property') {
+                return {
+                    ...op,
+                    _op: 'drop_property@ex1:0.1.1'
+                };
+            }
+            return op;
+        });
+        const { workers } = jobSpec;
+
+        const assetResponse = await terasliceHarness.teraslice.assets.getAsset('ex1', '0.0.1');
+        const assetId = assetResponse[0].id;
+
+        const ex = await terasliceHarness.submitAndStart(jobSpec);
+
+        const waitResponse = await terasliceHarness.forWorkersJoined(
+            ex.id(),
+            workers as number,
+            25
+        );
+        expect(waitResponse).toEqual(workers);
+
+        const execution = await ex.config();
+        expect(execution.assets[0]).toEqual(assetId);
+
+        await ex.stop({ blocking: true });
+    });
 });
 
 describe('s3 asset storage', () => {

packages/data-mate/package.json

@@ -31,7 +31,7 @@
     },
     "dependencies": {
         "@terascope/data-types": "~1.6.0",
-        "@terascope/types": "~1.3.1",
+        "@terascope/types": "~1.3.2",
         "@terascope/utils": "~1.6.0",
         "@types/validator": "~13.12.2",
         "awesome-phonenumber": "~7.2.0",

packages/data-types/package.json

@@ -27,7 +27,7 @@
         "test:watch": "ts-scripts test --watch . --"
     },
     "dependencies": {
-        "@terascope/types": "~1.3.1",
+        "@terascope/types": "~1.3.2",
         "@terascope/utils": "~1.6.0",
         "graphql": "~16.9.0",
         "yargs": "~17.7.2"

packages/elasticsearch-api/package.json

@@ -24,7 +24,7 @@
         "test:watch": "TEST_RESTRAINED_ELASTICSEARCH='true' ts-scripts test --watch . --"
     },
     "dependencies": {
-        "@terascope/types": "~1.3.1",
+        "@terascope/types": "~1.3.2",
         "@terascope/utils": "~1.6.0",
         "bluebird": "~3.7.2",
         "setimmediate": "~1.0.5"

packages/elasticsearch-store/package.json

@@ -32,7 +32,7 @@
     "dependencies": {
         "@terascope/data-mate": "~1.6.0",
         "@terascope/data-types": "~1.6.0",
-        "@terascope/types": "~1.3.1",
+        "@terascope/types": "~1.3.2",
         "@terascope/utils": "~1.6.0",
         "ajv": "~8.17.1",
         "ajv-formats": "~3.0.1",

packages/job-components/package.json

@@ -32,14 +32,15 @@
         "test:watch": "ts-scripts test --watch . --"
     },
     "dependencies": {
-        "@terascope/types": "~1.3.1",
+        "@terascope/types": "~1.3.2",
         "@terascope/utils": "~1.6.0",
         "convict": "~6.2.4",
         "convict-format-with-moment": "~6.2.0",
         "convict-format-with-validator": "~6.2.0",
         "datemath-parser": "~1.0.6",
         "import-meta-resolve": "~4.1.0",
         "prom-client": "~15.1.3",
+        "semver": "~7.6.3",
         "uuid": "~11.0.3"
     },
     "devDependencies": {

packages/job-components/src/execution-context/base.ts

@@ -43,8 +43,8 @@ export default class BaseExecutionContext<T extends OperationLifeCycle> {
         this.events.on('execution:add-to-lifecycle', this._handlers['execution:add-to-lifecycle']);
 
         const executionConfig = cloneDeep(config.executionConfig);
+
         this._loader = new OperationLoader({
-            terasliceOpPath: config.terasliceOpPath,
             assetPath: config.context.sysconfig.teraslice.assets_directory,
         });
 

packages/job-components/src/execution-context/interfaces.ts

@@ -9,7 +9,6 @@ import { APICore, OperationAPIType } from '../operations/index.js';
 export interface ExecutionContextConfig {
     context: Context;
     executionConfig: ExecutionConfig;
-    terasliceOpPath?: string;
     assetIds?: string[];
 }