Fix output stage w main merge.

README.md

@@ -88,10 +88,11 @@ For configuring allowed validators for verifying an asset signature before index
 export ALLOWED_VALIDATORS=[\"0x123\",\"0x456\"]
 ```
 
-For configuring a C2D (Compute to Data) cluster(s), please set the following environment variable (array of 1 or multiple cluster URLS):
+For configuring a C2D (Compute to Data) cluster(s), please set the following environment variable (array of 1 or multiple cluster URLS) and node URI that C2D points to, in order fetch the datasets & algorithms:
 
 ```bash
 export OPERATOR_SERVICE_URL=[\"http://example.c2d.cluster1.com\",\"http://example.cd2.cluster2.com\"]
+export C2D_NODE_URI='http://127.0.0.1:8081' #for e.g.
 ```
 
 For configuring the Indexer crawling interval in miliseconds (default, if not set, is 30 secs)

src/@types/OceanNode.ts

@@ -69,6 +69,7 @@ export interface OceanNodeConfig {
   supportedNetworks?: RPCS
   indexingNetworks?: RPCS
   c2dClusters: C2DClusterInfo[]
+  c2dNodeUri: string
   accountPurgatoryUrl: string
   assetPurgatoryUrl: string
   allowedAdmins?: string[]

src/components/c2d/compute_engines.ts

@@ -197,6 +197,7 @@ export class C2DEngineOPFK8 extends C2DEngine {
   ): Promise<ComputeJob[]> {
     // let's build the stage first
     // start with stage.input
+    const config = await getConfiguration()
     const stagesInput: OPFK8ComputeStageInput[] = []
     let index = 0
     for (const asset of assets) {
@@ -217,6 +218,14 @@ export class C2DEngineOPFK8 extends C2DEngine {
         })
       index++
     }
+    let getOuput = {}
+    if (output) {
+      getOuput = output
+    } else if (config.hasHttp && config.c2dNodeUri) {
+      getOuput = {
+        metadataUri: config.c2dNodeUri
+      }
+    }
     // continue with algorithm
     const stageAlgorithm: OPFK8ComputeStageAlgorithm = {}
     if (algorithm.url) {
@@ -237,7 +246,7 @@ export class C2DEngineOPFK8 extends C2DEngine {
       index: 0,
       input: stagesInput,
       algorithm: stageAlgorithm,
-      output: output || {},
+      output: getOuput,
       compute: {
         Instances: 1,
         namespace: environment,
@@ -250,7 +259,6 @@ export class C2DEngineOPFK8 extends C2DEngine {
     }
     // and the full payload
     const nonce: number = new Date().getTime()
-    const config = await getConfiguration()
     const providerSignature = await sign(String(nonce), config.keys.privateKey)
     const payload: OPFK8ComputeStart = {
       workflow,

src/test/integration/compute.test.ts

@@ -569,7 +569,8 @@ describe('Compute', () => {
         serviceId: publishedAlgoDataset.ddo.services[0].id,
         transferTxId: algoOrderTxId,
         meta: publishedAlgoDataset.ddo.metadata.algorithm
-      }
+      },
+      output: {}
       // additionalDatasets?: ComputeAsset[]
       // output?: ComputeOutput
     }

src/utils/config.ts

@@ -553,6 +553,7 @@ async function getEnvConfig(isStartup?: boolean): Promise<OceanNodeConfig> {
     indexingNetworks,
     feeStrategy: getOceanNodeFees(supportedNetworks, isStartup),
     c2dClusters: getC2DClusterEnvironment(isStartup),
+    c2dNodeUri: getEnvValue(process.env.C2D_NODE_URI, ''),
     accountPurgatoryUrl: getEnvValue(process.env.ACCOUNT_PURGATORY_URL, ''),
     assetPurgatoryUrl: getEnvValue(process.env.ASSET_PURGATORY_URL, ''),
     allowedAdmins: getAllowedAdmins(isStartup),