MVP of Logginghouse extension (0.2)

.github/workflows/build-and-publish-edc.yml

@@ -25,6 +25,9 @@ jobs:
         uses: gradle/gradle-build-action@749f47bda3e44aa060e82d7b3ef7e40d953bd629
         with:
           arguments: build 
+        env:
+          USERNAME: ${{ github.actor }}
+          TOKEN: ${{ github.token }}
       - name: Publish package
         if: ${{ github.ref == 'refs/heads/main' }}
         uses: gradle/gradle-build-action@749f47bda3e44aa060e82d7b3ef7e40d953bd629

.gitignore

@@ -1,5 +1,8 @@
 .DS_Store
 .idea/
+/**/.classpath
+/**/.settings
+/**/.project
 
 # Ignore Gradle project-specific cache directory
 .gradle

build.gradle.kts

@@ -41,6 +41,16 @@ allprojects {
     repositories {
         mavenCentral()
         mavenLocal()
+        repositories {
+            maven {
+                name = "GitHubPackages"
+                url = uri("https://maven.pkg.github.com/ids-basecamp/ids-infomodel-java")
+                credentials {
+                    username = System.getenv("USERNAME")
+                    password = System.getenv("TOKEN")
+                }
+            }
+        }
     }
 }
 

gradle.properties

@@ -5,7 +5,7 @@ assertj=3.23.1
 jupiterVersion=5.8.2
 mockitoVersion=4.8.0
 okHttpVersion=4.10.0
-jsonVersion=20230618
+jsonVersion=20231013
 jettyGroup=org.eclipse.jetty
 jettyVersion=11.0.15
 

logging-house-client/build.gradle.kts

@@ -13,9 +13,14 @@ val jsonVersion: String by project
 
 dependencies {
     implementation("${edcGroup}:control-plane-core:${edcVersion}")
+    implementation("${edcGroup}:http-spi:${edcVersion}")
 
     implementation("com.squareup.okhttp3:okhttp:${okHttpVersion}")
     implementation("org.json:json:${jsonVersion}")
+    implementation("org.glassfish.jersey.media:jersey-media-multipart:3.1.3")
+
+    implementation("de.fraunhofer.iais.eis.ids.infomodel:infomodel-java:1.0.2-basecamp")
+    implementation("de.fraunhofer.iais.eis.ids.infomodel:infomodel-util:1.0.2-basecamp")
 
     testImplementation("org.assertj:assertj-core:${assertj}")
     testImplementation("org.junit.jupiter:junit-jupiter-api:${jupiterVersion}")

logging-house-client/src/main/java/com/truzzt/extension/logginghouse/client/CreateProcessMessage.java

@@ -0,0 +1,39 @@
+/*
+ *  Copyright (c) 2024 truzzt GmbH
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       truzzt GmbH - initial API and implementation
+ *
+ */
+
+package com.truzzt.extension.logginghouse.client;
+
+import org.eclipse.edc.spi.types.domain.message.RemoteMessage;
+
+import java.net.URI;
+import java.net.URL;
+import java.util.List;
+
+public record CreateProcessMessage(
+        URL clearingHouseLogUrl,
+        URI connectorBaseUrl,
+        String processId,
+        List<String> processOwners
+) implements RemoteMessage {
+
+    @Override
+    public String getProtocol() {
+        return ExtendedMessageProtocolClearing.IDS_EXTENDED_PROTOCOL_CLEARING;
+    }
+
+    @Override
+    public String getCounterPartyAddress() {
+        return clearingHouseLogUrl.toString();
+    }
+}

logging-house-client/src/main/java/com/truzzt/extension/logginghouse/client/CreateProcessMessageSender.java

@@ -0,0 +1,69 @@
+/*
+ *  Copyright (c) 2024 truzzt GmbH
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       truzzt GmbH - initial API and implementation
+ *
+ */
+
+package com.truzzt.extension.logginghouse.client;
+
+import com.truzzt.extension.logginghouse.client.ids.jsonld.JsonLd;
+import com.truzzt.extension.logginghouse.client.ids.multipart.CalendarUtil;
+import com.truzzt.extension.logginghouse.client.ids.multipart.IdsConstants;
+import com.truzzt.extension.logginghouse.client.ids.multipart.IdsMultipartParts;
+import com.truzzt.extension.logginghouse.client.ids.multipart.MultipartResponse;
+import com.truzzt.extension.logginghouse.client.ids.multipart.MultipartSenderDelegate;
+import com.truzzt.extension.logginghouse.client.ids.multipart.ResponseUtil;
+import de.fraunhofer.iais.eis.DynamicAttributeToken;
+import de.fraunhofer.iais.eis.LogMessageBuilder;
+import de.fraunhofer.iais.eis.Message;
+import de.fraunhofer.iais.eis.MessageProcessedNotificationMessageImpl;
+import org.json.JSONObject;
+
+import java.util.List;
+
+public class CreateProcessMessageSender implements MultipartSenderDelegate<CreateProcessMessage, String> {
+
+    public CreateProcessMessageSender() {
+    }
+
+    @Override
+    public Message buildMessageHeader(CreateProcessMessage createProcessMessage, DynamicAttributeToken token) {
+        return new LogMessageBuilder()
+                ._modelVersion_(IdsConstants.INFORMATION_MODEL_VERSION)
+                ._issued_(CalendarUtil.gregorianNow())
+                ._securityToken_(token)
+                ._issuerConnector_(createProcessMessage.connectorBaseUrl())
+                ._senderAgent_(createProcessMessage.connectorBaseUrl())
+                .build();
+    }
+
+    @Override
+    public String buildMessagePayload(CreateProcessMessage createProcessMessage) {
+        var jo = new JSONObject();
+        jo.put("owners", createProcessMessage.processOwners());
+        return jo.toString();
+    }
+
+    @Override
+    public MultipartResponse<String> getResponseContent(IdsMultipartParts parts) throws Exception {
+        return ResponseUtil.parseMultipartStringResponse(parts, JsonLd.getObjectMapper());
+    }
+
+    @Override
+    public List<Class<? extends Message>> getAllowedResponseTypes() {
+        return List.of(MessageProcessedNotificationMessageImpl.class);
+    }
+
+    @Override
+    public Class<CreateProcessMessage> getMessageType() {
+        return CreateProcessMessage.class;
+    }
+}

logging-house-client/src/main/java/com/truzzt/extension/logginghouse/client/ExtendedMessageProtocolClearing.java

@@ -0,0 +1,25 @@
+/*
+ *  Copyright (c) 2022 sovity GmbH
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       sovity GmbH - initial API and implementation
+ *
+ */
+
+package com.truzzt.extension.logginghouse.client;
+
+public final class ExtendedMessageProtocolClearing {
+
+    private static final String EXTENDED_SUFFIX = "-extended-clearing";
+    public static final String IDS_MULTIPART = "ids-multipart";
+    public static final String IDS_EXTENDED_PROTOCOL_CLEARING = String.format("%s%s", IDS_MULTIPART, EXTENDED_SUFFIX);
+
+    private ExtendedMessageProtocolClearing() {
+    }
+}

logging-house-client/src/main/java/com/truzzt/extension/logginghouse/client/IdsClearingHouseServiceImpl.java

@@ -0,0 +1,126 @@
+/*
+ *  Copyright (c) 2022 sovity GmbH
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       sovity GmbH - initial API and implementation
+ *
+ */
+
+package com.truzzt.extension.logginghouse.client;
+
+import org.eclipse.edc.connector.contract.spi.event.contractnegotiation.ContractNegotiationFinalized;
+import org.eclipse.edc.connector.contract.spi.negotiation.store.ContractNegotiationStore;
+import org.eclipse.edc.connector.contract.spi.types.agreement.ContractAgreement;
+import org.eclipse.edc.connector.transfer.spi.event.TransferProcessTerminated;
+import org.eclipse.edc.connector.transfer.spi.store.TransferProcessStore;
+import org.eclipse.edc.connector.transfer.spi.types.TransferProcess;
+import org.eclipse.edc.spi.EdcException;
+import org.eclipse.edc.spi.event.Event;
+import org.eclipse.edc.spi.event.EventEnvelope;
+import org.eclipse.edc.spi.event.EventSubscriber;
+import org.eclipse.edc.spi.message.RemoteMessageDispatcherRegistry;
+import org.eclipse.edc.spi.monitor.Monitor;
+import org.eclipse.edc.spi.system.Hostname;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+public class IdsClearingHouseServiceImpl implements EventSubscriber {
+
+    private final RemoteMessageDispatcherRegistry dispatcherRegistry;
+    private final URI connectorBaseUrl;
+    private final URL clearingHouseLogUrl;
+    private final ContractNegotiationStore contractNegotiationStore;
+    private final TransferProcessStore transferProcessStore;
+    private final Monitor monitor;
+
+    public IdsClearingHouseServiceImpl(
+            RemoteMessageDispatcherRegistry dispatcherRegistry,
+            Hostname hostname,
+            URL clearingHouseLogUrl,
+            ContractNegotiationStore contractNegotiationStore,
+            TransferProcessStore transferProcessStore,
+            Monitor monitor) {
+        this.dispatcherRegistry = dispatcherRegistry;
+        this.clearingHouseLogUrl = clearingHouseLogUrl;
+        this.contractNegotiationStore = contractNegotiationStore;
+        this.transferProcessStore = transferProcessStore;
+        this.monitor = monitor;
+
+        try {
+            connectorBaseUrl = getConnectorBaseUrl(hostname);
+        } catch (URISyntaxException e) {
+            throw new EdcException("Could not create connectorBaseUrl. Hostname can be set using:" +
+                    " edc.hostname", e);
+        }
+    }
+
+    public void createProcess(ContractAgreement contractAgreement, URL clearingHouseLogUrl) {
+        // Create PID
+        List<String> processOwners = new ArrayList<>();
+        processOwners.add(contractAgreement.getConsumerId());
+        processOwners.add(contractAgreement.getProviderId());
+
+        monitor.info("Creating Process in LoggingHouse");
+        var logMessage = new CreateProcessMessage(clearingHouseLogUrl, connectorBaseUrl, contractAgreement.getId(), processOwners);
+        dispatcherRegistry.dispatch(Object.class, logMessage);
+    }
+
+    public void logContractAgreement(ContractAgreement contractAgreement, URL clearingHouseLogUrl) {
+        monitor.info("Logging contract agreement to LoggingHouse");
+        var logMessage = new LogMessage(clearingHouseLogUrl, connectorBaseUrl, contractAgreement);
+        dispatcherRegistry.dispatch(Object.class, logMessage);
+    }
+
+    public void logTransferProcess(TransferProcess transferProcess, URL clearingHouseLogUrl) {
+        monitor.info("Logging transferprocess to LoggingHouse");
+        var logMessage = new LogMessage(clearingHouseLogUrl, connectorBaseUrl, transferProcess);
+        dispatcherRegistry.dispatch(Object.class, logMessage);
+    }
+
+    @Override
+    public <E extends Event> void on(EventEnvelope<E> event) {
+        try {
+            if (event.getPayload() instanceof ContractNegotiationFinalized contractNegotiationFinalized) {
+                var contractAgreement = resolveContractAgreement(contractNegotiationFinalized);
+                var pid = contractAgreement.getId();
+                var extendedUrl = new URL(clearingHouseLogUrl + "/" + pid);
+
+                createProcess(contractAgreement, clearingHouseLogUrl);
+                logContractAgreement(contractAgreement, extendedUrl);
+            } else if (event.getPayload() instanceof TransferProcessTerminated transferProcessTerminated) {
+                var transferProcess = resolveTransferProcess(transferProcessTerminated);
+                var pid = transferProcess.getContractId();
+                var extendedUrl = new URL(clearingHouseLogUrl + "/" + pid);
+                logTransferProcess(transferProcess, extendedUrl);
+            }
+        } catch (Exception e) {
+            throw new EdcException("Could not create extended clearinghouse url.");
+        }
+    }
+
+    private ContractAgreement resolveContractAgreement(ContractNegotiationFinalized contractNegotiationFinalized) throws NullPointerException {
+        var contractNegotiationId = contractNegotiationFinalized.getContractNegotiationId();
+        var contractNegotiation = contractNegotiationStore.findById(contractNegotiationId);
+        return Objects.requireNonNull(contractNegotiation).getContractAgreement();
+    }
+
+    private TransferProcess resolveTransferProcess(TransferProcessTerminated transferProcessTerminated) {
+        var transferProcessId = transferProcessTerminated.getTransferProcessId();
+        return transferProcessStore.findById(transferProcessId);
+    }
+
+    private URI getConnectorBaseUrl(Hostname hostname) throws URISyntaxException {
+        return new URI(String.format("https://%s/", hostname.get()));
+    }
+}

logging-house-client/src/main/java/com/truzzt/extension/logginghouse/client/IdsMultipartClearingRemoteMessageDispatcher.java

@@ -0,0 +1,30 @@
+/*
+ *  Copyright (c) 2022 sovity GmbH
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       sovity GmbH - initial API and implementation
+ *
+ */
+
+package com.truzzt.extension.logginghouse.client;
+
+import com.truzzt.extension.logginghouse.client.ids.multipart.IdsMultipartRemoteMessageDispatcher;
+import com.truzzt.extension.logginghouse.client.ids.multipart.IdsMultipartSender;
+
+public class IdsMultipartClearingRemoteMessageDispatcher extends IdsMultipartRemoteMessageDispatcher {
+
+    public IdsMultipartClearingRemoteMessageDispatcher(IdsMultipartSender idsMultipartSender) {
+        super(idsMultipartSender);
+    }
+
+    @Override
+    public String protocol() {
+        return ExtendedMessageProtocolClearing.IDS_EXTENDED_PROTOCOL_CLEARING;
+    }
+}

logging-house-client/src/main/java/com/truzzt/extension/logginghouse/client/LogMessage.java

@@ -0,0 +1,34 @@
+/*
+ *  Copyright (c) 2022 sovity GmbH
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       sovity GmbH - initial API and implementation
+ *
+ */
+
+package com.truzzt.extension.logginghouse.client;
+
+import org.eclipse.edc.spi.types.domain.message.RemoteMessage;
+
+import java.net.URI;
+import java.net.URL;
+
+public record LogMessage(URL clearingHouseLogUrl,
+                         URI connectorBaseUrl,
+                         Object eventToLog) implements RemoteMessage {
+    @Override
+    public String getProtocol() {
+        return ExtendedMessageProtocolClearing.IDS_EXTENDED_PROTOCOL_CLEARING;
+    }
+
+    @Override
+    public String getCounterPartyAddress() {
+        return clearingHouseLogUrl.toString();
+    }
+}