HTTP Data Plane Support

[gonzalogil26]

Instead of Blob storage container provisioning, I think support for HTTP data plane is of gret interest for the MVD. How can it be supported. What must be included in the Connector build.gradle.kts?

[paullatzelsperger]

hey @gonzalogil26 I've converted this into a discussion, because that's were these kinds of inquiries are best kept.

I am not quite certain as to the nature of your question: are you suggesting we should add a test track for the HTTP dataplane? Or do you want to set up your own MVD using the HTTP data plane?

[gonzalogil26]

I have already the MVD running. To implement it in a real use case, I would like to set up the MVD using the HTTP data plance, so that I can expose data from a REST API.

[paullatzelsperger]

you will certainly need the data-plane-http module on your classpath, I suggest you read the module documentation.

Generally though the PipelineServiceImpl will evaluate the DataFlowRequest, and use the HttpDataSourceFactory to generate an (ephemeral) HttpDataSource, which will serve as the pipeline's datasource. Same thing for the sink, HTTP or other.

[oliverbusch]

Hello @gonzalogil26, have you implemented the use case in the meantime?

[oliverbusch]

Hello @paullatzelsperger,
we've a problem to integrate the transfer-06-consumer-pull-http sample into the MVD. The behavior of the Dashboard UI looks fine, but the data transfer doesn't work.

Attached our steps and some source codes extracts:

1. We've created a new rest-connector directory in the launcher directory with following files

   • build.gradle.kts
   • Dockerfile

2. We've changed the docker-compose.yml in the system-test directory

   • docker-compose.yml

3. We've created a new rest-connector directory in the root directory of the MVD. The directory has the following structure

   • A rest-consumer directory
     • build.gradle.kts
     • config.properties
   • A rest-provider directory
     • build.gradle.kts
     • config.properties
     • src directory with the Extension

       import jakarta.ws.rs.core.MediaType;
       import java.util.UUID;
       import org.eclipse.edc.connector.contract.spi.offer.store.ContractDefinitionStore;
       import org.eclipse.edc.connector.contract.spi.types.offer.ContractDefinition;
       import org.eclipse.edc.connector.dataplane.spi.pipeline.DataTransferExecutorServiceContainer;
       import org.eclipse.edc.connector.dataplane.selector.spi.instance.DataPlaneInstance;
       import org.eclipse.edc.connector.dataplane.selector.spi.store.DataPlaneInstanceStore;
       import org.eclipse.edc.connector.policy.spi.PolicyDefinition;
       import org.eclipse.edc.connector.policy.spi.store.PolicyDefinitionStore;
       import org.eclipse.edc.policy.model.Action;
       import org.eclipse.edc.policy.model.Permission;
       import org.eclipse.edc.policy.model.Policy;
       import org.eclipse.edc.runtime.metamodel.annotation.Extension;
       import org.eclipse.edc.runtime.metamodel.annotation.Inject;
       import org.eclipse.edc.spi.asset.AssetIndex;
       import org.eclipse.edc.spi.asset.AssetSelectorExpression;
       import org.eclipse.edc.spi.monitor.Monitor;
       import org.eclipse.edc.spi.system.ServiceExtension;
       import org.eclipse.edc.spi.system.ServiceExtensionContext;
       import org.eclipse.edc.spi.system.health.HealthCheckService;
       import org.eclipse.edc.spi.types.domain.HttpDataAddress;
       import org.eclipse.edc.spi.types.domain.asset.Asset;
       import org.eclipse.edc.web.spi.WebService;
       
       @Extension(value = RestProviderApiExtension.NAME)
       public class RestProviderApiExtension implements ServiceExtension {
        private Monitor monitor;
        public static final String NAME = "RestProviderApiExtension";
        public static final String USE_POLICY = "use-eu";
        private static final String EDC_ASSET_PATH = "data";
        private static final String EDC_ASSET_METHOD = "GET";
        
       @Inject
        private WebService webService;
        @Inject
        private ContractDefinitionStore contractStore;
        @Inject
        private HealthCheckService healthService;
        @Inject
        private AssetIndex assetIndex;
        @Inject
        private DataTransferExecutorServiceContainer executorContainer;
        @Inject
        private PolicyDefinitionStore policyStore;
        @Inject 
        
        private DataPlaneInstanceStore dataplaneStore;
        
        @Override
        public String name() {
        return NAME;
        }
        
        @Override
        public void initialize(ServiceExtensionContext context) {
        monitor = context.getMonitor();
        
       if (isProvider(context)) {
        registerDataplaneProvider(context, "http-pull-provider-dataplane"); 
       registerAsset(context);
        var policy = noConstraintPolicy();
        policyStore.save(policy);
        registerContractDefinition(policy.getUid());
        context.registerService(WebService.class, webService); 
       webService.registerResource(new RestProviderApiController(context.getMonitor()));
        } else {
        registerDataplaneConsumer(context, "http-pull-consumer-dataplane");
        }
        context.registerService(HealthCheckService.class, healthService);
        }
        
       private PolicyDefinition noConstraintPolicy() {
        return PolicyDefinition.Builder
        .newInstance().id(UUID.randomUUID().toString()).policy(
        Policy.Builder.newInstance()
        .permission(Permission.Builder.newInstance()
        .action(Action.Builder.newInstance().type("USE").build()).build())
        .build())
        .build();
        }
        
       private void registerDataplaneProvider(ServiceExtensionContext context, String uid) {
        monitor.info("T4T_Provider");
        var dataplane = DataPlaneInstance.Builder.newInstance()
        .id(uid)
        .url("http://localhost:19192/control/transfer")
        .allowedSourceType("HttpData")
        .allowedDestType("HttpProxy")
        .allowedDestType("HttpData")
        .property("publicApiUrl", "http://localhost:19291/public/")
        .build();
        dataplaneStore.create(dataplane);
        }
        
       private void registerDataplaneConsumer(ServiceExtensionContext context, String uid) {
        monitor.info("T4T_Consumer");
        var dataplane = DataPlaneInstance.Builder.newInstance()
        .id(uid)
        .url("http://localhost:29192/control/transfer")
        .allowedSourceType("HttpData")
        .allowedDestType("HttpProxy")
        .allowedDestType("HttpData")
        .property("publicApiUrl", "http://localhost:29291/public/")
        .build();
        dataplaneStore.create(dataplane);
        }
        
       private void registerAsset(ServiceExtensionContext context) {
        //String assetBaseUrlSetting = "http://localhost:8181/api/provider";
        //String assetBaseUrlSetting = "http://company1:8181/api/provider";
        String assetBaseUrlSetting = "http://jsonplaceholder.typicode.com/users";
        String assetPathSetting = context.getSetting(EDC_ASSET_PATH, "data");
        String assetMethodSetting = context.getSetting(EDC_ASSET_METHOD, "GET");
        HttpDataAddress dataAddress = HttpDataAddress.Builder.newInstance().type(HttpDataAddress.HTTP_DATA)
        .property("name", "Test asset")
        .property("baseUrl", assetBaseUrlSetting)
        .property("type", HttpDataAddress.HTTP_DATA)
        //.property("path", assetPathSetting)
        //.property("method", assetMethodSetting)
        .addAdditionalHeader("x-api-key", "ApiKeyDefaultValue")
        .keyName("HttpData")
        .build();
        
       String assetId = "assetId";
        String assetName = "product description";
        String assetContenttype = MediaType.APPLICATION_JSON; 
       String assetVersion = "1.0";
        Asset asset = Asset.Builder.newInstance().id(assetId).name(assetName).description(assetName).version(assetVersion).contentType(assetContenttype)
        .property("type", HttpDataAddress.HTTP_DATA)
        .build();
        
       assetIndex.accept(asset, dataAddress);
        }
        private void registerContractDefinition(String id) {
        var contractDefinition = ContractDefinition.Builder.newInstance().id(id).accessPolicyId(id)
        .contractPolicyId(id)
        .selectorExpression(
        AssetSelectorExpression.Builder.newInstance().whenEquals(Asset.PROPERTY_ID, "assetId").build())
        .validity(31536000) // valid for a year
        .build();
        contractStore.save(contractDefinition);
        }
        
       private boolean isProvider(ServiceExtensionContext context) {
        monitor.info("edc.identity.did.url" + context.getConfig().getString("edc.identity.did.url"));
        boolean result = true;
        
       if (context.getConfig().getString("edc.identity.did.url").equals("did:web:did-server:company1")) {
        result = true;
        } else {
        result = false;
        }
        
       return result;
        }
       }`

     • A rest-transfer directory
       • build.gradle.kts

4. We've added the folowing entries to the settings.gradle.kts file

   • library("data-plane-api", "org.eclipse.edc", "data-plane-api").versionRef("edc")
   • library("data-plane-core", "org.eclipse.edc", "data-plane-core").versionRef("edc")
   • library("data-plane-http", "org.eclipse.edc", "data-plane-http").versionRef("edc")
   • library("spi-core", "org.eclipse.edc", "core-spi").versionRef("edc")
   • library("spi-transfer-data-plane", "org.eclipse.edc", "transfer-data-plane-spi").versionRef("edc")
   • library("api-management", "org.eclipse.edc", "management-api").versionRef("edc")
   • library("ext-transfer-pull-http-receiver", "org.eclipse.edc", "transfer-pull-http-receiver").versionRef("edc")
   • library("data-plane-client", "org.eclipse.edc", "data-plane-client").versionRef("edc")

5. We've extended the dashboard UI

   • app.module.ts

       {
           provide: 'STORAGE_TYPES',
           useFactory: () => [{id: "AzureStorage", name: "AzureStorage"}, {id: "AmazonS3", name: "AmazonS3"}, {id: "HttpProxy", name: "HttpProxy"}],
       },

   • contract-viewer.component.ts

     private createTransferRequest(contract: ContractAgreementDto, storageTypeId: string): Observable<TransferRequestDto> {
     if (storageTypeId == 'HttpProxy') {
       return this.getOfferedAssetForId(contract.assetId!).pipe(map(offeredAsset => {
         return {
           assetId: offeredAsset.id,
           contractId: contract.id,
           connectorId: "http-pull-provider",
           dataDestination: {
             properties: {
               "type": "HttpProxy"
             }
           },
           managedResources: false,
           transferType: {isFinite: true}, //must be there, otherwise NPE on backend
           connectorAddress: offeredAsset.originator,
           protocol: 'ids-multipart'
         };
       }));
     }
     else {
     return this.getOfferedAssetForId(contract.assetId!).pipe(map(offeredAsset => {
       return {
         assetId: offeredAsset.id,
         contractId: contract.id,
         connectorId: "consumer", //doesn't matter, but cannot be null
         dataDestination: {
           properties: {
             "type": storageTypeId,
             account: this.homeConnectorStorageAccount, // CAUTION: hardcoded value for AzureBlob
             // container: omitted, so it will be auto-assigned by the EDC runtime
           }
         },
         managedResources: true,
         transferType: {isFinite: true}, //must be there, otherwise NPE on backend
         connectorAddress: offeredAsset.originator,
         protocol: 'ids-multipart'
       };
     }));
     }	  

   • transfer-history-viewer.component.ts

     showStorageExplorerLink(transferProcess: TransferProcessDto) {
     return (transferProcess.dataDestination?.properties?.type === 'AzureStorage' ||
     transferProcess.dataDestination?.properties?.type === 'HttpProxy')
     && transferProcess.state === 'COMPLETED';
     }
         

6. With this changes we can build and start the MVD and the transfer-06-consumer-pull-http Backend-Service.

7. Via the Dashboard we can now do the steps out of the transfer-06-consumer-pull-http README.md

   • Negotiate a contract
   • Getting the contract agreement id
   • Start the transfer
   • Check the transfer status

   In the logfiles everything looks fine, but the command

    `curl --location --request GET 'http://localhost:29291/public/' --header 'Authorization: <auth code>'`
   

   doesn't calls the external rest-service and delivers not results.

   Any idea, any hint?

[oliverbusch]

We've found a solution for our Problem.

[AHABID]

is there any possibility to know more about the solution please ?

[nimritee]

Hello, is there a chance for a solution? I'm currently working on a university project that involves the same implementation, and any assistance would be greatly appreciated.

[oliverbusch]

dd_changes.zip
Attached the changes we've done in the DataDashboard repository

[gboldrini]

Hi, I'm also trying to add the http-data-plane inside my MVD, I've added all the necessary dependencies that I found in the Samples reository and I also modified the UI. The MVD work properly (I'm able to upload a file without problem) but the Http transfer doesn't work. So I've got a question: "it's necessary to add the http-request-logger server in order to transfer data via HttpProxy or I'm missing sono environment variables? "
Thank you in advance