This is an automated email from the ASF dual-hosted git repository. SvenO3 pushed a commit to branch 5193-fix-missing-icons-and-documentation-during-extension-migration in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 18da8b3ebaaf7c2c6e920b3accbb94e08bcf2d75 Author: Sven Oehler <[email protected]> AuthorDate: Tue Nov 3 11:28:21 2026 +0100 Ensure to get documentation from correct service --- .../streampipes/manager/assets/AssetFetcher.java | 5 + .../streampipes/manager/assets/AssetManager.java | 23 +++- .../migration/AbstractMigrationManager.java | 6 +- .../manager/verification/ElementVerifier.java | 16 ++- .../manager/verification/TypedElementVerifier.java | 5 +- .../migration/MigrationAssetUpdateTest.java | 132 +++++++++++++++++++++ 6 files changed, 171 insertions(+), 16 deletions(-) diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java index 08e28c0ede..36b438fe0c 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java @@ -22,6 +22,7 @@ import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestMana import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestTargets; import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequests; import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator; +import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration; import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery; import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider; @@ -49,6 +50,10 @@ public class AssetFetcher { public InputStream fetchPipelineElementAssets() throws IOException, NoServiceEndpointsAvailableException { var service = new ExtensionsServiceEndpointGenerator(serviceDiscovery).selectService(appId, spServiceUrlProvider, Set.of()); + return fetchPipelineElementAssets(service); + } + + public InputStream fetchPipelineElementAssets(SpServiceRegistration service) throws IOException { var requestTarget = ExtensionServiceRequestTargets.pipelineElementAssets(service, spServiceUrlProvider, appId); var response = requestManager.request(ExtensionServiceRequests.pipelineElementAssets(requestTarget)); diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetManager.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetManager.java index 525aa67a73..68d231bccd 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetManager.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetManager.java @@ -20,6 +20,7 @@ package org.apache.streampipes.manager.assets; import org.apache.streampipes.commons.constants.GlobalStreamPipesConstants; import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException; import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager; +import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration; import org.apache.streampipes.storage.api.system.ISpCoreConfigurationStorage; import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery; import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider; @@ -59,12 +60,22 @@ public class AssetManager { } public void storeAsset(SpServiceUrlProvider spServiceUrlProvider, - String appId, - ExtensionServiceRequestManager requestManager, - ISpServiceDiscovery serviceDiscovery) throws IOException, NoServiceEndpointsAvailableException { - InputStream assetStream = new AssetFetcher(serviceDiscovery, spServiceUrlProvider, appId, requestManager) - .fetchPipelineElementAssets(); - new AssetExtractor(assetStream, appId, coreConfigurationStorage).extractAssetContents(); + String appId, + ExtensionServiceRequestManager requestManager, + ISpServiceDiscovery serviceDiscovery) throws IOException, NoServiceEndpointsAvailableException { + storeAsset(spServiceUrlProvider, appId, requestManager, serviceDiscovery, null); + } + + public void storeAsset(SpServiceUrlProvider spServiceUrlProvider, + String appId, + ExtensionServiceRequestManager requestManager, + ISpServiceDiscovery serviceDiscovery, + SpServiceRegistration service) throws IOException, NoServiceEndpointsAvailableException { + var fetcher = new AssetFetcher(serviceDiscovery, spServiceUrlProvider, appId, requestManager); + try (InputStream assetStream = service == null + ? fetcher.fetchPipelineElementAssets() : fetcher.fetchPipelineElementAssets(service)) { + new AssetExtractor(assetStream, appId, coreConfigurationStorage).extractAssetContents(); + } } public void deleteAsset(String appId) throws IOException { diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AbstractMigrationManager.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AbstractMigrationManager.java index c933c3aefd..5afbdbaf4f 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AbstractMigrationManager.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AbstractMigrationManager.java @@ -148,21 +148,21 @@ public abstract class AbstractMigrationManager { .forEach(config -> { if (isInstalled(config.modelType(), config.targetAppId())) { var requestTarget = getRequestTarget(config.modelType(), config.targetAppId(), service); - performUpdate(requestTarget); + performUpdate(requestTarget, service); } }); } protected abstract boolean isInstalled(SpServiceTagPrefix modelType, String appId); - protected void performUpdate(ExtensionServiceRequestTarget requestTarget) { + protected void performUpdate(ExtensionServiceRequestTarget requestTarget, SpServiceRegistration service) { try { var entityPayload = requestManager .request(ExtensionServiceRequests.descriptionUpdate(requestTarget, resourceManager)) .responseBody(); var updateResult = new TypeExtractor(entityPayload, descriptionStorage, requestManager, resourceManager) - .getTypeVerifier().verifyAndUpdate(); + .getTypeVerifier().verifyAndUpdate(service); if (!updateResult.isSuccess()) { LOG.error( "Updating the pipeline element description failed: {}", diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/ElementVerifier.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/ElementVerifier.java index b18fdb3c84..125c9d3c9b 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/ElementVerifier.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/ElementVerifier.java @@ -23,6 +23,7 @@ import org.apache.streampipes.manager.assets.AssetManager; import org.apache.streampipes.model.base.NamedStreamPipesEntity; import org.apache.streampipes.model.client.user.Permission; import org.apache.streampipes.model.client.user.PermissionBuilder; +import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration; import org.apache.streampipes.model.message.ErrorMessage; import org.apache.streampipes.model.message.Message; import org.apache.streampipes.model.message.Notification; @@ -96,7 +97,7 @@ public abstract class ElementVerifier<T extends NamedStreamPipesEntity> { if (state == StorageState.STORED) { createAndStorePermission(principalSid, publicElement); try { - storeAssets(); + storeAssets(null); } catch (IOException | NoServiceEndpointsAvailableException e) { LOG.error("Could not store assets for app id '{}'", elementDescription.getAppId(), e); } @@ -108,6 +109,10 @@ public abstract class ElementVerifier<T extends NamedStreamPipesEntity> { } public Message verifyAndUpdate() { + return verifyAndUpdate(null); + } + + public Message verifyAndUpdate(SpServiceRegistration service) { var transformError = transformEntity(); if (transformError != null) { return transformError; @@ -115,7 +120,7 @@ public abstract class ElementVerifier<T extends NamedStreamPipesEntity> { update(); try { - updateAssets(); + updateAssets(service); } catch (IOException | NoServiceEndpointsAvailableException e) { LOG.error("Could not update assets for app id '{}'", elementDescription.getAppId(), e); } @@ -123,12 +128,13 @@ public abstract class ElementVerifier<T extends NamedStreamPipesEntity> { } - protected abstract void storeAssets() throws IOException, NoServiceEndpointsAvailableException; + protected abstract void storeAssets(SpServiceRegistration service) + throws IOException, NoServiceEndpointsAvailableException; - protected void updateAssets() throws IOException, NoServiceEndpointsAvailableException { + protected void updateAssets(SpServiceRegistration service) throws IOException, NoServiceEndpointsAvailableException { if (elementDescription.isIncludesAssets()) { assetManager.deleteAsset(elementDescription.getAppId()); - storeAssets(); + storeAssets(service); } } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/TypedElementVerifier.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/TypedElementVerifier.java index b8bb1ac341..7ec16fdc67 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/TypedElementVerifier.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/TypedElementVerifier.java @@ -21,6 +21,7 @@ package org.apache.streampipes.manager.verification; import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException; import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager; import org.apache.streampipes.model.base.NamedStreamPipesEntity; +import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration; import org.apache.streampipes.resource.management.SpResourceManager; import org.apache.streampipes.storage.api.pipeline.IPipelineElementDescriptionStorage; import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery; @@ -93,9 +94,9 @@ public class TypedElementVerifier<T extends NamedStreamPipesEntity> extends Elem } @Override - protected void storeAssets() throws IOException, NoServiceEndpointsAvailableException { + protected void storeAssets(SpServiceRegistration service) throws IOException, NoServiceEndpointsAvailableException { if (elementDescription.isIncludesAssets()) { - assetManager.storeAsset(serviceUrlProvider, elementDescription.getAppId(), requestManager, serviceDiscovery); + assetManager.storeAsset(serviceUrlProvider, elementDescription.getAppId(), requestManager, serviceDiscovery, service); } } } diff --git a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/migration/MigrationAssetUpdateTest.java b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/migration/MigrationAssetUpdateTest.java new file mode 100644 index 0000000000..fe7fb4ca13 --- /dev/null +++ b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/migration/MigrationAssetUpdateTest.java @@ -0,0 +1,132 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.apache.streampipes.manager.migration; + +import org.apache.streampipes.manager.api.extensions.ExtensionServiceOperationResult; +import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager; +import org.apache.streampipes.manager.util.AuthTokenProvider; +import org.apache.streampipes.model.client.user.Principal; +import org.apache.streampipes.model.configuration.SpCoreConfiguration; +import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration; +import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix; +import org.apache.streampipes.model.extensions.transport.ExtensionServiceBrokerOperations; +import org.apache.streampipes.model.graph.DataSinkDescription; +import org.apache.streampipes.model.migration.ModelMigratorConfig; +import org.apache.streampipes.resource.management.SpResourceManager; +import org.apache.streampipes.resource.management.UserResourceManager; +import org.apache.streampipes.storage.api.pipeline.IPipelineElementDescriptionStorage; +import org.apache.streampipes.storage.api.system.ISpCoreConfigurationStorage; +import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.zip.ZipEntry; +import java.util.zip.ZipOutputStream; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +class MigrationAssetUpdateTest { + + private static final String APP_ID = "org.apache.streampipes.sinks.internal.jvm.dataset"; + + @TempDir + Path directory; + + private SpResourceManager resources; + private IPipelineElementDescriptionStorage descriptions; + private ISpServiceDiscovery discovery; + private SpServiceRegistration service; + + @BeforeEach + void setUp() { + resources = mock(SpResourceManager.class); + descriptions = mock(IPipelineElementDescriptionStorage.class); + discovery = mock(ISpServiceDiscovery.class); + when(resources.getServiceDiscovery()).thenReturn(discovery); + var configurationStorage = mock(ISpCoreConfigurationStorage.class); + var configuration = new SpCoreConfiguration(); + configuration.setAssetDir(directory.toString()); + when(configurationStorage.get()).thenReturn(configuration); + when(resources.getCoreConfigurationStorage()).thenReturn(configurationStorage); + service = new SpServiceRegistration(); + service.setSvcId("streampipes-current-instance"); + service.setHost("localhost"); + service.setPort(8090); + } + + @Test + void migrationFetchesAssetsFromTheServiceProvidingTheDescription() throws Exception { + var archive = assetArchive(); + ExtensionServiceRequestManager requests = request -> { + assertEquals(service.getSvcId(), request.target().serviceId()); + if (request.target().operation().equals(ExtensionServiceBrokerOperations.PIPELINE_ELEMENT_ASSETS.operationId())) { + return new ExtensionServiceOperationResult(200, archive); + } + return new ExtensionServiceOperationResult(200, description().getBytes(StandardCharsets.UTF_8)); + }; + var manager = new AbstractMigrationManager(requests, resources, descriptions) { + @Override + protected boolean isInstalled(SpServiceTagPrefix modelType, String appId) { + return true; + } + }; + + var users = mock(UserResourceManager.class); + when(resources.manageUsers()).thenReturn(users); + when(users.getServiceAdmin()).thenReturn(mock(Principal.class)); + try (var tokens = mockConstruction(AuthTokenProvider.class)) { + manager.updateDescriptions(List.of(new ModelMigratorConfig(APP_ID, SpServiceTagPrefix.DATA_SINK, 1, 2)), service); + } + + assertEquals("new icon", Files.readString(directory.resolve(APP_ID).resolve("icon.png"))); + verify(descriptions).update(any(DataSinkDescription.class)); + verifyNoInteractions(discovery); + } + + private String description() { + return "{\"@class\":\"" + DataSinkDescription.class.getCanonicalName() + + "\",\"appId\":\"" + APP_ID + "\",\"elementId\":\"test-element\",\"includesAssets\":true}"; + } + + private byte[] assetArchive() throws IOException { + var bytes = new ByteArrayOutputStream(); + try (var zip = new ZipOutputStream(bytes)) { + zip.putNextEntry(new ZipEntry("icon.png")); + zip.write("new icon".getBytes(StandardCharsets.UTF_8)); + zip.closeEntry(); + zip.putNextEntry(new ZipEntry("documentation.md")); + zip.write("documentation".getBytes(StandardCharsets.UTF_8)); + zip.closeEntry(); + } + return bytes.toByteArray(); + } +}
