This is an automated email from the ASF dual-hosted git repository. dominikriemer pushed a commit to branch add-running-instance-dialog-registered-services in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit a62c555e43ca9185228aa3ffb35dc9021e980b9a Author: Dominik Riemer <[email protected]> AuthorDate: Wed Jul 8 09:38:44 2026 +0200 feat: Show running instances in service settings --- .../ExtensionInstanceDetailsProvider.java | 170 ++++++++ .../ExtensionInstanceRemovalService.java | 183 +++++++++ .../monitoring/model/RunningAdapterInstance.java | 28 ++ .../model/RunningExtensionInstances.java | 26 ++ .../model/RunningPipelineElementInstance.java | 28 ++ .../impl/admin/ServiceRegistrationResource.java | 77 ++++ ...xtensions-service-details-dialog.component.html | 2 +- ...service-running-instances-dialog.component.html | 454 +++++++++++++++++++++ ...service-running-instances-dialog.component.scss | 48 +++ ...s-service-running-instances-dialog.component.ts | 371 +++++++++++++++++ .../registered-extensions-services.component.html | 16 +- .../registered-extensions-services.component.ts | 17 +- .../configuration/shared/configuration.service.ts | 66 +++ 13 files changed, 1480 insertions(+), 6 deletions(-) diff --git a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ExtensionInstanceDetailsProvider.java b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ExtensionInstanceDetailsProvider.java new file mode 100644 index 0000000000..8aef254bec --- /dev/null +++ b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ExtensionInstanceDetailsProvider.java @@ -0,0 +1,170 @@ +/* + * 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.health.monitoring; + +import org.apache.streampipes.commons.constants.InstanceIdExtractor; +import org.apache.streampipes.health.monitoring.model.ActiveResources; +import org.apache.streampipes.health.monitoring.model.RunningAdapterInstance; +import org.apache.streampipes.health.monitoring.model.RunningExtensionInstances; +import org.apache.streampipes.health.monitoring.model.RunningPipelineElementInstance; +import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager; +import org.apache.streampipes.model.base.InvocableStreamPipesEntity; +import org.apache.streampipes.model.connect.adapter.AdapterDescription; +import org.apache.streampipes.model.graph.DataProcessorInvocation; +import org.apache.streampipes.model.graph.DataSinkInvocation; +import org.apache.streampipes.model.health.ExtensionInstanceHealth; +import org.apache.streampipes.model.pipeline.Pipeline; +import org.apache.streampipes.resource.management.SpResourceManager; +import org.apache.streampipes.storage.api.system.IExtensionsServiceStorage; + +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +public class ExtensionInstanceDetailsProvider { + + private final IExtensionsServiceStorage extensionsServiceStorage; + private final ResourceProvider resourceProvider; + private final ExtensionServiceRequestManager extensionRequestManager; + private final SpResourceManager resourceManager; + + public ExtensionInstanceDetailsProvider(IExtensionsServiceStorage extensionsServiceStorage, + ResourceProvider resourceProvider, + ExtensionServiceRequestManager extensionRequestManager, + SpResourceManager resourceManager) { + this.extensionsServiceStorage = extensionsServiceStorage; + this.resourceProvider = resourceProvider; + this.extensionRequestManager = extensionRequestManager; + this.resourceManager = resourceManager; + } + + public RunningExtensionInstances getRunningInstances(String serviceId) { + var extensionHealth = new ExtensionInstanceAvailabilityCheck( + extensionsServiceStorage, + serviceId, + extensionRequestManager, + resourceManager + ).checkRunningInstances(); + + var activeResources = resourceProvider.loadActiveResources(); + return new RunningExtensionInstances( + serviceId, + getRunningAdapters(serviceId, extensionHealth, activeResources), + getRunningPipelineElements(serviceId, extensionHealth, activeResources) + ); + } + + private List<RunningAdapterInstance> getRunningAdapters(String serviceId, + ExtensionInstanceHealth extensionHealth, + ActiveResources activeResources) { + Map<String, AdapterDescription> adaptersByInstanceId = activeResources.allAdapters() + .stream() + .filter(adapter -> serviceId.equals(adapter.getSelectedServiceId())) + .collect(Collectors.toMap( + AdapterDescription::getElementId, + adapter -> adapter, + (existing, replacement) -> existing + )); + + return extensionHealth.adapterInstanceStates() + .entrySet() + .stream() + .map(entry -> { + var adapter = adaptersByInstanceId.get(entry.getKey()); + return new RunningAdapterInstance( + entry.getKey(), + adapter != null ? adapter.getName() : null, + adapter != null ? adapter.getAppId() : null, + entry.getValue(), + adapter == null + ); + }) + .sorted(Comparator + .comparing(RunningAdapterInstance::orphaned).reversed() + .thenComparing(RunningAdapterInstance::name, + Comparator.nullsLast(String.CASE_INSENSITIVE_ORDER)) + .thenComparing(RunningAdapterInstance::instanceId)) + .toList(); + } + + private List<RunningPipelineElementInstance> getRunningPipelineElements(String serviceId, + ExtensionInstanceHealth extensionHealth, + ActiveResources activeResources) { + Map<String, RunningPipelineElementInstance> pipelineElementsByInstanceId = activeResources.runningPipelines() + .stream() + .flatMap(pipeline -> runningPipelineElements(pipeline) + .filter(element -> serviceId.equals(element.getSelectedServiceId())) + .map(element -> new RunningPipelineElementInstance( + InstanceIdExtractor.extractId(element.getElementId()), + element.getName(), + element.getAppId(), + pipeline.getPipelineId(), + pipeline.getName(), + getPipelineElementType(element), + false + ))) + .collect(Collectors.toMap( + RunningPipelineElementInstance::instanceId, + element -> element, + (existing, replacement) -> existing + )); + + return extensionHealth.runningPipelineElementInstanceIds() + .stream() + .map(instanceId -> pipelineElementsByInstanceId.getOrDefault( + instanceId, + new RunningPipelineElementInstance( + instanceId, + null, + null, + null, + null, + "UNKNOWN", + true + ) + )) + .sorted(Comparator + .comparing(RunningPipelineElementInstance::orphaned).reversed() + .thenComparing(RunningPipelineElementInstance::pipelineName, + Comparator.nullsLast(String.CASE_INSENSITIVE_ORDER)) + .thenComparing(RunningPipelineElementInstance::name, + Comparator.nullsLast(String.CASE_INSENSITIVE_ORDER)) + .thenComparing(RunningPipelineElementInstance::instanceId)) + .toList(); + } + + private Stream<InvocableStreamPipesEntity> runningPipelineElements(Pipeline pipeline) { + return Stream.concat( + pipeline.getSepas().stream(), + pipeline.getActions().stream() + ); + } + + private String getPipelineElementType(InvocableStreamPipesEntity element) { + if (element instanceof DataProcessorInvocation) { + return "PROCESSOR"; + } else if (element instanceof DataSinkInvocation) { + return "SINK"; + } else { + return "UNKNOWN"; + } + } +} diff --git a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ExtensionInstanceRemovalService.java b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ExtensionInstanceRemovalService.java new file mode 100644 index 0000000000..1fa612fbc8 --- /dev/null +++ b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ExtensionInstanceRemovalService.java @@ -0,0 +1,183 @@ +/* + * 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.health.monitoring; + +import org.apache.streampipes.commons.constants.InstanceIdExtractor; +import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager; +import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestTargets; +import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequests; +import org.apache.streampipes.model.base.InvocableStreamPipesEntity; +import org.apache.streampipes.model.connect.adapter.AdapterDescription; +import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration; +import org.apache.streampipes.model.graph.DataSinkInvocation; +import org.apache.streampipes.model.pipeline.Pipeline; +import org.apache.streampipes.resource.management.SpResourceManager; +import org.apache.streampipes.serializers.json.JacksonSerializer; +import org.apache.streampipes.storage.api.system.IExtensionsServiceStorage; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider; + +import com.fasterxml.jackson.core.JsonProcessingException; + +import java.io.IOException; +import java.util.Objects; +import java.util.stream.Stream; + +public class ExtensionInstanceRemovalService { + + private final IExtensionsServiceStorage extensionsServiceStorage; + private final ResourceProvider resourceProvider; + private final ExtensionServiceRequestManager extensionRequestManager; + private final SpResourceManager resourceManager; + + public ExtensionInstanceRemovalService(IExtensionsServiceStorage extensionsServiceStorage, + ResourceProvider resourceProvider, + ExtensionServiceRequestManager extensionRequestManager, + SpResourceManager resourceManager) { + this.extensionsServiceStorage = extensionsServiceStorage; + this.resourceProvider = resourceProvider; + this.extensionRequestManager = extensionRequestManager; + this.resourceManager = resourceManager; + } + + public void removeAllInstances(String serviceId) throws IOException { + var runningInstances = new ExtensionInstanceDetailsProvider( + extensionsServiceStorage, + resourceProvider, + extensionRequestManager, + resourceManager + ).getRunningInstances(serviceId); + + for (var adapter : runningInstances.adapters()) { + removeAdapterInstance(serviceId, adapter.instanceId()); + } + + for (var pipelineElement : runningInstances.pipelineElements()) { + removePipelineElementInstance(serviceId, pipelineElement.instanceId()); + } + } + + public void removeAdapterInstance(String serviceId, + String instanceId) throws IOException { + var service = getService(serviceId); + var activeResources = resourceProvider.loadActiveResources(); + var adapter = activeResources.allAdapters() + .stream() + .filter(candidate -> serviceId.equals(candidate.getSelectedServiceId())) + .filter(candidate -> instanceId.equals(candidate.getElementId())) + .findFirst() + .map(AdapterDescription::new) + .orElseGet(() -> makeOrphanedAdapterDescription(instanceId)); + + adapter.setRunning(true); + + var requestTarget = ExtensionServiceRequestTargets.adapterStop(service); + var response = extensionRequestManager.request( + ExtensionServiceRequests.adapterStateChange( + requestTarget, + instanceId, + serialize(adapter), + resourceManager + ) + ); + + if (!response.isSuccess()) { + throw new IOException("Could not remove adapter instance %s from service %s: %s" + .formatted(instanceId, serviceId, response.responseBody())); + } + } + + public void removePipelineElementInstance(String serviceId, + String instanceId) throws IOException { + var service = getService(serviceId); + var activeResources = resourceProvider.loadActiveResources(); + var match = activeResources.runningPipelines() + .stream() + .flatMap(pipeline -> runningPipelineElements(pipeline) + .filter(element -> serviceId.equals(element.getSelectedServiceId())) + .filter(element -> instanceId.equals(InstanceIdExtractor.extractId(element.getElementId()))) + .map(element -> new PipelineElementRemovalTarget( + element.getAppId(), + pipeline.getPipelineId(), + getProvider(element) + ))) + .filter(Objects::nonNull) + .findFirst() + .orElseGet(() -> new PipelineElementRemovalTarget( + instanceId, + null, + SpServiceUrlProvider.DATA_PROCESSOR + )); + + var requestTarget = ExtensionServiceRequestTargets.pipelineDetach( + service.getServiceUrl(), + service.getSvcId(), + match.provider(), + match.appId(), + instanceId + ); + var response = extensionRequestManager.request( + ExtensionServiceRequests.pipelineElementDetach( + requestTarget, + match.pipelineId(), + resourceManager + ) + ); + + if (!response.isSuccess()) { + throw new IOException("Could not remove pipeline element instance %s from service %s: %s" + .formatted(instanceId, serviceId, response.responseBody())); + } + } + + private SpServiceRegistration getService(String serviceId) throws IOException { + return extensionsServiceStorage.findAll() + .stream() + .filter(service -> serviceId.equals(service.getSvcId())) + .findFirst() + .orElseThrow(() -> new IOException("Could not find extension service " + serviceId)); + } + + private AdapterDescription makeOrphanedAdapterDescription(String instanceId) { + var adapterDescription = new AdapterDescription(); + adapterDescription.setElementId(instanceId); + return adapterDescription; + } + + private String serialize(AdapterDescription adapterDescription) throws JsonProcessingException { + return JacksonSerializer.getObjectMapper().writeValueAsString(adapterDescription); + } + + private Stream<InvocableStreamPipesEntity> runningPipelineElements(Pipeline pipeline) { + return Stream.concat( + pipeline.getSepas().stream(), + pipeline.getActions().stream() + ); + } + + private SpServiceUrlProvider getProvider(InvocableStreamPipesEntity element) { + return element instanceof DataSinkInvocation + ? SpServiceUrlProvider.DATA_SINK + : SpServiceUrlProvider.DATA_PROCESSOR; + } + + private record PipelineElementRemovalTarget(String appId, + String pipelineId, + SpServiceUrlProvider provider) { + } +} diff --git a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/RunningAdapterInstance.java b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/RunningAdapterInstance.java new file mode 100644 index 0000000000..1f90ab3da1 --- /dev/null +++ b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/RunningAdapterInstance.java @@ -0,0 +1,28 @@ +/* + * 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.health.monitoring.model; + +import org.apache.streampipes.model.health.AdapterInstanceState; + +public record RunningAdapterInstance(String instanceId, + String name, + String appId, + AdapterInstanceState state, + boolean orphaned) { +} diff --git a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/RunningExtensionInstances.java b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/RunningExtensionInstances.java new file mode 100644 index 0000000000..007f6fd0f2 --- /dev/null +++ b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/RunningExtensionInstances.java @@ -0,0 +1,26 @@ +/* + * 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.health.monitoring.model; + +import java.util.List; + +public record RunningExtensionInstances(String serviceId, + List<RunningAdapterInstance> adapters, + List<RunningPipelineElementInstance> pipelineElements) { +} diff --git a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/RunningPipelineElementInstance.java b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/RunningPipelineElementInstance.java new file mode 100644 index 0000000000..4a46ba5ae1 --- /dev/null +++ b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/RunningPipelineElementInstance.java @@ -0,0 +1,28 @@ +/* + * 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.health.monitoring.model; + +public record RunningPipelineElementInstance(String instanceId, + String name, + String appId, + String pipelineId, + String pipelineName, + String type, + boolean orphaned) { +} diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java index ce19d8b6db..eadb100e3d 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java @@ -17,10 +17,16 @@ */ package org.apache.streampipes.rest.impl.admin; +import org.apache.streampipes.health.monitoring.ExtensionInstanceDetailsProvider; +import org.apache.streampipes.health.monitoring.ExtensionInstanceRemovalService; +import org.apache.streampipes.health.monitoring.ResourceProvider; import org.apache.streampipes.health.monitoring.ServiceRegistrationManager; +import org.apache.streampipes.health.monitoring.model.RunningExtensionInstances; +import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager; import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration; import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceStatus; import org.apache.streampipes.model.message.Notifications; +import org.apache.streampipes.resource.management.SpResourceManager; import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource; import org.apache.streampipes.rest.security.AuthConstants; import org.apache.streampipes.rest.shared.exception.SpMessageException; @@ -30,6 +36,7 @@ import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; import org.springframework.security.access.prepost.PreAuthorize; +import org.springframework.web.bind.annotation.DeleteMapping; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.PostMapping; @@ -37,6 +44,7 @@ import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; +import java.io.IOException; import java.util.List; @RestController @@ -46,12 +54,64 @@ public class ServiceRegistrationResource extends AbstractAuthGuardedRestResource private final IExtensionsServiceStorage extensionsServiceStorage = getNoSqlStorage().getExtensionsServiceStorage(); + private final ExtensionServiceRequestManager extensionServiceRequestManager; + private final SpResourceManager resourceManager; + + public ServiceRegistrationResource(ExtensionServiceRequestManager extensionServiceRequestManager, + SpResourceManager resourceManager) { + this.extensionServiceRequestManager = extensionServiceRequestManager; + this.resourceManager = resourceManager; + } @GetMapping(produces = MediaType.APPLICATION_JSON_VALUE) public ResponseEntity<List<SpServiceRegistration>> getRegisteredServices() { return ok(extensionsServiceStorage.findAll()); } + @GetMapping(path = "/{serviceId}/running-instances", produces = MediaType.APPLICATION_JSON_VALUE) + public ResponseEntity<RunningExtensionInstances> getRunningInstances(@PathVariable("serviceId") String serviceId) { + return ok(new ExtensionInstanceDetailsProvider( + extensionsServiceStorage, + makeResourceProvider(), + extensionServiceRequestManager, + resourceManager + ).getRunningInstances(serviceId)); + } + + @DeleteMapping(path = "/{serviceId}/running-instances", produces = MediaType.APPLICATION_JSON_VALUE) + public ResponseEntity<Void> removeAllRunningInstances(@PathVariable("serviceId") String serviceId) { + try { + makeRemovalService().removeAllInstances(serviceId); + return ok(); + } catch (IOException e) { + throw new SpMessageException(HttpStatus.BAD_REQUEST, Notifications.error(e.getMessage())); + } + } + + @DeleteMapping(path = "/{serviceId}/running-instances/adapters/{instanceId}", + produces = MediaType.APPLICATION_JSON_VALUE) + public ResponseEntity<Void> removeAdapterInstance(@PathVariable("serviceId") String serviceId, + @PathVariable("instanceId") String instanceId) { + try { + makeRemovalService().removeAdapterInstance(serviceId, instanceId); + return ok(); + } catch (IOException e) { + throw new SpMessageException(HttpStatus.BAD_REQUEST, Notifications.error(e.getMessage())); + } + } + + @DeleteMapping(path = "/{serviceId}/running-instances/pipeline-elements/{instanceId}", + produces = MediaType.APPLICATION_JSON_VALUE) + public ResponseEntity<Void> removePipelineElementInstance(@PathVariable("serviceId") String serviceId, + @PathVariable("instanceId") String instanceId) { + try { + makeRemovalService().removePipelineElementInstance(serviceId, instanceId); + return ok(); + } catch (IOException e) { + throw new SpMessageException(HttpStatus.BAD_REQUEST, Notifications.error(e.getMessage())); + } + } + @PostMapping(consumes = MediaType.APPLICATION_JSON_VALUE) public ResponseEntity<Void> registerService(@RequestBody SpServiceRegistration serviceRegistration) { new ServiceRegistrationManager(extensionsServiceStorage).addService(serviceRegistration, @@ -69,4 +129,21 @@ public class ServiceRegistrationResource extends AbstractAuthGuardedRestResource Notifications.error("Could not find registered service with id " + serviceId)); } } + + private ResourceProvider makeResourceProvider() { + return new ResourceProvider( + resourceManager.managePipelines().getDb(), + resourceManager.manageAdapters().getDb(), + null + ); + } + + private ExtensionInstanceRemovalService makeRemovalService() { + return new ExtensionInstanceRemovalService( + extensionsServiceStorage, + makeResourceProvider(), + extensionServiceRequestManager, + resourceManager + ); + } } diff --git a/ui/src/app/configuration/dialog/extensions-service-details/extensions-service-details-dialog.component.html b/ui/src/app/configuration/dialog/extensions-service-details/extensions-service-details-dialog.component.html index 8122b4f195..007a1c8414 100644 --- a/ui/src/app/configuration/dialog/extensions-service-details/extensions-service-details-dialog.component.html +++ b/ui/src/app/configuration/dialog/extensions-service-details/extensions-service-details-dialog.component.html @@ -55,7 +55,7 @@ </div> </div> <mat-divider></mat-divider> - <div class="sp-dialog-actions actions-align-right"> + <div class="sp-dialog-actions"> <button mat-button mat-flat-button class="mat-basic" (click)="close()"> {{ 'Close' | translate }} </button> diff --git a/ui/src/app/configuration/dialog/extensions-service-running-instances/extensions-service-running-instances-dialog.component.html b/ui/src/app/configuration/dialog/extensions-service-running-instances/extensions-service-running-instances-dialog.component.html new file mode 100644 index 0000000000..3253a9b4e9 --- /dev/null +++ b/ui/src/app/configuration/dialog/extensions-service-running-instances/extensions-service-running-instances-dialog.component.html @@ -0,0 +1,454 @@ +<!-- +~ 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. +~ +--> + +<div class="sp-dialog-container"> + <div class="sp-dialog-content p-15"> + <div fxLayout="column" fxLayoutGap="20px" class="p-15"> + <div fxLayout="row" fxLayoutAlign="space-between start"> + <div fxLayout="column" fxLayoutGap="5px"> + <div class="text-md"> + {{ serviceReg.svcId }} + </div> + <div class="text-sm"> + {{ serviceReg.svcGroup }} + </div> + </div> + <div> + <button + mat-icon-button + color="accent" + [disabled]="loading || removing" + [matTooltip]="'Refresh running instances' | translate" + (click)="loadRunningInstances()" + > + <mat-icon>refresh</mat-icon> + </button> + </div> + </div> + + @if (loading) { + <div + fxLayout="row" + fxLayoutAlign="center center" + class="loading-container" + > + <mat-spinner diameter="32"></mat-spinner> + </div> + } + + @if (!loading && loadError) { + <div class="empty-state"> + {{ 'Could not load running instances' | translate }} + </div> + } + + @if (!loading && !loadError) { + <div fxLayout="column" fxLayoutGap="10px"> + <div class="section-title"> + {{ 'Adapters' | translate }} + </div> + @if (adapterDataSource.data.length > 0) { + <sp-table + fxFlex="100" + [dataSource]="adapterDataSource" + [columns]="adapterColumns" + [nameSearchConfig]="adapterNameSearchConfig" + > + <ng-container matColumnDef="name"> + <th + fxFlex="25" + fxLayoutAlign="start center" + mat-header-cell + *matHeaderCellDef + > + {{ 'Name' | translate }} + </th> + <td + fxFlex="25" + fxLayoutAlign="start center" + mat-cell + *matCellDef="let adapter" + > + {{ adapter.displayName | translate }} + </td> + </ng-container> + + <ng-container matColumnDef="state"> + <th + fxFlex="15" + fxLayoutAlign="start center" + mat-header-cell + *matHeaderCellDef + > + {{ 'State' | translate }} + </th> + <td + fxFlex="15" + fxLayoutAlign="start center" + mat-cell + *matCellDef="let adapter" + > + <sp-label + size="small" + variant="soft" + [tone]=" + getAdapterStateTone(adapter.state) + " + [labelText]="adapter.state | translate" + > + </sp-label> + </td> + </ng-container> + + <ng-container matColumnDef="relation"> + <th + fxFlex="15" + fxLayoutAlign="start center" + mat-header-cell + *matHeaderCellDef + > + {{ 'Relation' | translate }} + </th> + <td + fxFlex="15" + fxLayoutAlign="start center" + mat-cell + *matCellDef="let adapter" + > + <sp-label + size="small" + variant="soft" + [tone]=" + getRelationTone(adapter.orphaned) + " + [labelText]=" + adapter.relation | translate + " + > + </sp-label> + </td> + </ng-container> + + <ng-container matColumnDef="instanceId"> + <th + fxFlex="35" + fxLayoutAlign="start center" + mat-header-cell + *matHeaderCellDef + > + {{ 'Instance ID' | translate }} + </th> + <td + fxFlex="35" + fxLayoutAlign="start center" + mat-cell + *matCellDef="let adapter" + class="instance-id" + > + {{ adapter.instanceId }} + </td> + </ng-container> + + <ng-container matColumnDef="action"> + <th + fxFlex="10" + fxLayoutAlign="end center" + mat-header-cell + *matHeaderCellDef + ></th> + <td + fxFlex="10" + fxLayoutAlign="end center" + mat-cell + *matCellDef="let adapter" + > + <button + mat-icon-button + class="btn-warn" + [disabled]="removing" + [matTooltip]=" + 'Remove instance from service' + | translate + " + (click)=" + requestRemoveAdapterInstance( + adapter + ); + $event.stopPropagation() + " + > + <mat-icon>delete</mat-icon> + </button> + </td> + </ng-container> + </sp-table> + } @else { + <div class="empty-state"> + {{ 'No running adapters' | translate }} + </div> + } + </div> + + <div fxLayout="column" fxLayoutGap="10px"> + <div class="section-title"> + {{ 'Pipeline elements' | translate }} + </div> + @if (pipelineElementDataSource.data.length > 0) { + <sp-table + fxFlex="100" + [dataSource]="pipelineElementDataSource" + [columns]="pipelineElementColumns" + [nameSearchConfig]="pipelineElementNameSearchConfig" + > + <ng-container matColumnDef="name"> + <th + fxFlex="20" + fxLayoutAlign="start center" + mat-header-cell + *matHeaderCellDef + > + {{ 'Name' | translate }} + </th> + <td + fxFlex="20" + fxLayoutAlign="start center" + mat-cell + *matCellDef="let element" + > + {{ element.displayName | translate }} + </td> + </ng-container> + + <ng-container matColumnDef="pipelineName"> + <th + fxFlex="20" + fxLayoutAlign="start center" + mat-header-cell + *matHeaderCellDef + > + {{ 'Pipeline' | translate }} + </th> + <td + fxFlex="20" + fxLayoutAlign="start center" + mat-cell + *matCellDef="let element" + > + {{ element.pipelineDisplayName }} + </td> + </ng-container> + + <ng-container matColumnDef="type"> + <th + fxFlex="12" + fxLayoutAlign="start center" + mat-header-cell + *matHeaderCellDef + > + {{ 'Type' | translate }} + </th> + <td + fxFlex="12" + fxLayoutAlign="start center" + mat-cell + *matCellDef="let element" + > + <sp-label + size="small" + variant="soft" + [tone]=" + getPipelineElementTypeTone( + element.type + ) + " + [labelText]="element.type | translate" + > + </sp-label> + </td> + </ng-container> + + <ng-container matColumnDef="relation"> + <th + fxFlex="13" + fxLayoutAlign="start center" + mat-header-cell + *matHeaderCellDef + > + {{ 'Relation' | translate }} + </th> + <td + fxFlex="13" + fxLayoutAlign="start center" + mat-cell + *matCellDef="let element" + > + <sp-label + size="small" + variant="soft" + [tone]=" + getRelationTone(element.orphaned) + " + [labelText]=" + element.relation | translate + " + > + </sp-label> + </td> + </ng-container> + + <ng-container matColumnDef="instanceId"> + <th + fxFlex="25" + fxLayoutAlign="start center" + mat-header-cell + *matHeaderCellDef + > + {{ 'Instance ID' | translate }} + </th> + <td + fxFlex="25" + fxLayoutAlign="start center" + mat-cell + *matCellDef="let element" + class="instance-id" + > + {{ element.instanceId }} + </td> + </ng-container> + + <ng-container matColumnDef="action"> + <th + fxFlex="10" + fxLayoutAlign="end center" + mat-header-cell + *matHeaderCellDef + ></th> + <td + fxFlex="10" + fxLayoutAlign="end center" + mat-cell + *matCellDef="let element" + > + <button + mat-icon-button + color="accent" + [disabled]="removing" + [matTooltip]=" + 'Remove instance from service' + | translate + " + (click)=" + requestRemovePipelineElementInstance( + element + ); + $event.stopPropagation() + " + > + <mat-icon>delete</mat-icon> + </button> + </td> + </ng-container> + </sp-table> + } @else { + <div class="empty-state"> + {{ 'No running pipeline elements' | translate }} + </div> + } + </div> + } + </div> + </div> + <mat-divider></mat-divider> + <div + class="sp-dialog-actions running-instances-actions" + fxLayout="column" + fxLayoutGap="10px" + > + @if (pendingRemoval) { + <sp-alert-banner + class="confirmation-alert" + type="warning" + icon="warning" + [title]="getPendingRemovalTitle() | translate" + [description]=" + 'Deleted instances of active pipelines or adapters will be redeployed by the health check.' + | translate + " + > + <div + class="confirmation-content" + fxLayout="column" + fxLayoutAlign="start start" + fxLayoutGap="10px" + > + <div class="text-sm instance-id"> + {{ getPendingRemovalDescription() | translate }} + </div> + <div + fxLayout="row" + fxLayoutAlign="start center" + fxLayoutGap="10px" + > + <button + mat-button + mat-flat-button + class="mat-basic" + [disabled]="removing" + (click)="cancelRemoval()" + > + {{ 'Cancel' | translate }} + </button> + <button + mat-button + mat-flat-button + color="accent" + [disabled]="removing" + (click)="confirmRemoval()" + > + {{ 'Remove' | translate }} + </button> + </div> + </div> + </sp-alert-banner> + } + <div fxLayout="row" fxLayoutAlign="start center" fxLayoutGap="10px"> + <button + mat-button + mat-flat-button + class="btn-warn" + [disabled]=" + removing || + loading || + totalInstanceCount === 0 || + !!pendingRemoval + " + (click)="requestRemoveAllInstances()" + > + {{ 'Remove all instances' | translate }} + </button> + <button + mat-button + mat-flat-button + class="mat-basic" + (click)="close()" + > + {{ 'Close' | translate }} + </button> + </div> + </div> +</div> diff --git a/ui/src/app/configuration/dialog/extensions-service-running-instances/extensions-service-running-instances-dialog.component.scss b/ui/src/app/configuration/dialog/extensions-service-running-instances/extensions-service-running-instances-dialog.component.scss new file mode 100644 index 0000000000..18d4777d19 --- /dev/null +++ b/ui/src/app/configuration/dialog/extensions-service-running-instances/extensions-service-running-instances-dialog.component.scss @@ -0,0 +1,48 @@ +/*! + * 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. + * + */ + +.loading-container { + min-height: 120px; +} + +.section-title { + font-weight: 600; +} + +.empty-state { + color: var(--color-secondary-text); + padding: 10px 0; +} + +.confirmation-content { + margin-top: 8px; +} + +.confirmation-alert { + display: block; + width: 100%; +} + +.running-instances-actions { + align-items: stretch; + justify-content: flex-start; +} + +.instance-id { + overflow-wrap: anywhere; +} diff --git a/ui/src/app/configuration/dialog/extensions-service-running-instances/extensions-service-running-instances-dialog.component.ts b/ui/src/app/configuration/dialog/extensions-service-running-instances/extensions-service-running-instances-dialog.component.ts new file mode 100644 index 0000000000..c9b619b67a --- /dev/null +++ b/ui/src/app/configuration/dialog/extensions-service-running-instances/extensions-service-running-instances-dialog.component.ts @@ -0,0 +1,371 @@ +/* + * 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. + * + */ + +import { Component, Input, OnInit, inject } from '@angular/core'; +import { MatButton, MatIconButton } from '@angular/material/button'; +import { MatDivider } from '@angular/material/divider'; +import { MatIcon } from '@angular/material/icon'; +import { MatProgressSpinner } from '@angular/material/progress-spinner'; +import { + MatCell, + MatCellDef, + MatColumnDef, + MatHeaderCell, + MatHeaderCellDef, + MatTableDataSource, +} from '@angular/material/table'; +import { MatTooltip } from '@angular/material/tooltip'; +import { TranslatePipe } from '@ngx-translate/core'; +import { Observable } from 'rxjs'; +import { + FlexDirective, + LayoutAlignDirective, + LayoutDirective, + LayoutGapDirective, +} from '@ngbracket/ngx-layout/flex'; +import { SpServiceRegistration } from '@streampipes/platform-services'; +import { + DialogRef, + SpAlertBannerComponent, + SpLabelComponent, + SpTableComponent, + SpTableNameSearchConfig, +} from '@streampipes/shared-ui'; +import { + ConfigurationService, + RunningAdapterInstance, + RunningExtensionInstances, + RunningPipelineElementInstance, +} from '../../shared/configuration.service'; + +type RunningAdapterInstanceRow = RunningAdapterInstance & { + displayName: string; + relation: string; +}; + +type RunningPipelineElementInstanceRow = RunningPipelineElementInstance & { + displayName: string; + pipelineDisplayName: string; + relation: string; +}; + +type PendingRemoval = + | { + type: 'adapter'; + item: RunningAdapterInstanceRow; + } + | { + type: 'pipelineElement'; + item: RunningPipelineElementInstanceRow; + } + | { + type: 'all'; + }; + +@Component({ + selector: 'sp-extensions-service-running-instances-dialog', + templateUrl: './extensions-service-running-instances-dialog.component.html', + styleUrls: ['./extensions-service-running-instances-dialog.component.scss'], + imports: [ + FlexDirective, + LayoutAlignDirective, + LayoutDirective, + LayoutGapDirective, + MatButton, + MatCell, + MatCellDef, + MatColumnDef, + MatDivider, + MatHeaderCell, + MatHeaderCellDef, + MatIcon, + MatIconButton, + MatProgressSpinner, + MatTooltip, + SpAlertBannerComponent, + SpLabelComponent, + SpTableComponent, + TranslatePipe, + ], +}) +export class SpExtensionsServiceRunningInstancesDialogComponent implements OnInit { + private dialogRef = + inject<DialogRef<SpExtensionsServiceRunningInstancesDialogComponent>>( + DialogRef, + ); + private configurationService = inject(ConfigurationService); + + @Input() + serviceReg: SpServiceRegistration; + + adapterColumns = ['name', 'state', 'relation', 'instanceId', 'action']; + pipelineElementColumns = [ + 'name', + 'pipelineName', + 'type', + 'relation', + 'instanceId', + 'action', + ]; + runningInstances: RunningExtensionInstances; + adapterDataSource = new MatTableDataSource<RunningAdapterInstanceRow>(); + pipelineElementDataSource = + new MatTableDataSource<RunningPipelineElementInstanceRow>(); + adapterNameSearchConfig: SpTableNameSearchConfig<RunningAdapterInstanceRow> = + { + enabled: true, + placeholder: 'Filter adapters', + searchKey: [ + 'displayName', + 'instanceId', + 'appId', + 'state', + 'relation', + ], + }; + pipelineElementNameSearchConfig: SpTableNameSearchConfig<RunningPipelineElementInstanceRow> = + { + enabled: true, + placeholder: 'Filter pipeline elements', + searchKey: [ + 'displayName', + 'pipelineDisplayName', + 'instanceId', + 'appId', + 'pipelineId', + 'type', + 'relation', + ], + }; + loading = true; + loadError = false; + removing = false; + pendingRemoval?: PendingRemoval; + + get totalInstanceCount(): number { + return ( + this.adapterDataSource.data.length + + this.pipelineElementDataSource.data.length + ); + } + + ngOnInit(): void { + this.loadRunningInstances(); + } + + loadRunningInstances(): void { + this.loading = true; + this.loadError = false; + this.configurationService + .getRunningExtensionInstances(this.serviceReg.svcId) + .subscribe({ + next: response => { + this.runningInstances = response; + this.adapterDataSource.data = response.adapters.map( + adapter => this.makeAdapterRow(adapter), + ); + this.pipelineElementDataSource.data = + response.pipelineElements.map(pipelineElement => + this.makePipelineElementRow(pipelineElement), + ); + this.loading = false; + }, + error: () => { + this.loading = false; + this.loadError = true; + }, + }); + } + + requestRemoveAdapterInstance(adapter: RunningAdapterInstanceRow): void { + this.pendingRemoval = { + type: 'adapter', + item: adapter, + }; + } + + requestRemovePipelineElementInstance( + pipelineElement: RunningPipelineElementInstanceRow, + ): void { + this.pendingRemoval = { + type: 'pipelineElement', + item: pipelineElement, + }; + } + + requestRemoveAllInstances(): void { + this.pendingRemoval = { + type: 'all', + }; + } + + cancelRemoval(): void { + this.pendingRemoval = undefined; + } + + confirmRemoval(): void { + if (!this.pendingRemoval) { + return; + } + + if (this.pendingRemoval.type === 'adapter') { + this.removeAdapterInstance(this.pendingRemoval.item); + } else if (this.pendingRemoval.type === 'pipelineElement') { + this.removePipelineElementInstance(this.pendingRemoval.item); + } else { + this.removeAllInstances(); + } + } + + getPendingRemovalTitle(): string { + if (!this.pendingRemoval) { + return ''; + } + + if (this.pendingRemoval.type === 'adapter') { + return 'Remove adapter instance?'; + } else if (this.pendingRemoval.type === 'pipelineElement') { + return 'Remove pipeline element instance?'; + } else { + return 'Remove all running instances?'; + } + } + + getPendingRemovalDescription(): string { + if (!this.pendingRemoval) { + return ''; + } + + if (this.pendingRemoval.type === 'adapter') { + return this.pendingRemoval.item.instanceId; + } else if (this.pendingRemoval.type === 'pipelineElement') { + return this.pendingRemoval.item.instanceId; + } else { + return 'This removes all instances currently reported by this extension service.'; + } + } + + private removeAdapterInstance(adapter: RunningAdapterInstance): void { + this.removeInstance(() => + this.configurationService.removeRunningAdapterInstance( + this.serviceReg.svcId, + adapter.instanceId, + ), + ); + } + + private removePipelineElementInstance( + pipelineElement: RunningPipelineElementInstance, + ): void { + this.removeInstance(() => + this.configurationService.removeRunningPipelineElementInstance( + this.serviceReg.svcId, + pipelineElement.instanceId, + ), + ); + } + + private removeAllInstances(): void { + this.removeInstance(() => + this.configurationService.removeAllRunningExtensionInstances( + this.serviceReg.svcId, + ), + ); + } + + close(): void { + this.dialogRef.close(); + } + + getAdapterStateTone( + state: RunningAdapterInstance['state'], + ): 'success' | 'warning' | 'info' | 'neutral' { + if (state === 'RUNNING') { + return 'success'; + } else if (state === 'STOPPING') { + return 'warning'; + } else if (state === 'STARTING') { + return 'info'; + } else { + return 'neutral'; + } + } + + getPipelineElementTypeTone( + type: RunningPipelineElementInstance['type'], + ): 'info' | 'neutral' | 'warning' { + if (type === 'PROCESSOR') { + return 'info'; + } else if (type === 'SINK') { + return 'neutral'; + } else { + return 'warning'; + } + } + + getRelationLabel(orphaned: boolean): string { + return orphaned ? 'Orphaned' : 'Configured'; + } + + getRelationTone(orphaned: boolean): 'success' | 'warning' { + return orphaned ? 'warning' : 'success'; + } + + private makeAdapterRow( + adapter: RunningAdapterInstance, + ): RunningAdapterInstanceRow { + return { + ...adapter, + displayName: adapter.orphaned + ? 'Orphaned adapter instance' + : (adapter.name ?? ''), + relation: this.getRelationLabel(adapter.orphaned), + }; + } + + private makePipelineElementRow( + pipelineElement: RunningPipelineElementInstance, + ): RunningPipelineElementInstanceRow { + return { + ...pipelineElement, + displayName: pipelineElement.orphaned + ? 'Orphaned pipeline element instance' + : (pipelineElement.name ?? ''), + pipelineDisplayName: pipelineElement.orphaned + ? '-' + : (pipelineElement.pipelineName ?? ''), + relation: this.getRelationLabel(pipelineElement.orphaned), + }; + } + + private removeInstance(removeFn: () => Observable<object>): void { + this.removing = true; + this.loadError = false; + removeFn().subscribe({ + next: () => { + this.removing = false; + this.pendingRemoval = undefined; + this.loadRunningInstances(); + }, + error: () => { + this.removing = false; + this.loadError = true; + }, + }); + } +} diff --git a/ui/src/app/configuration/extensions-service-management/registered-extensions-services/registered-extensions-services.component.html b/ui/src/app/configuration/extensions-service-management/registered-extensions-services/registered-extensions-services.component.html index 61c79ce624..b4dc934aa2 100644 --- a/ui/src/app/configuration/extensions-service-management/registered-extensions-services/registered-extensions-services.component.html +++ b/ui/src/app/configuration/extensions-service-management/registered-extensions-services/registered-extensions-services.component.html @@ -53,7 +53,7 @@ <ng-container matColumnDef="group"> <th - fxFlex="45" + fxFlex="40" fxLayoutAlign="start center" mat-header-cell *matHeaderCellDef @@ -61,7 +61,7 @@ {{ 'Service Group' | translate }} </th> <td - fxFlex="45" + fxFlex="40" fxLayoutAlign="start center" mat-cell *matCellDef="let element" @@ -91,7 +91,7 @@ <ng-container matColumnDef="action"> <th - fxFlex="10" + fxFlex="15" fxLayoutAlign="end center" mat-header-cell *matHeaderCellDef @@ -106,11 +106,19 @@ </button> </th> <td - fxFlex="10" + fxFlex="15" fxLayoutAlign="end center" mat-cell *matCellDef="let element" > + <button + mat-icon-button + color="accent" + [matTooltip]="'View running instances' | translate" + (click)="openRunningInstances(element)" + > + <mat-icon>list_alt</mat-icon> + </button> <button mat-icon-button color="accent" diff --git a/ui/src/app/configuration/extensions-service-management/registered-extensions-services/registered-extensions-services.component.ts b/ui/src/app/configuration/extensions-service-management/registered-extensions-services/registered-extensions-services.component.ts index 9492e12313..785c15d1d7 100644 --- a/ui/src/app/configuration/extensions-service-management/registered-extensions-services/registered-extensions-services.component.ts +++ b/ui/src/app/configuration/extensions-service-management/registered-extensions-services/registered-extensions-services.component.ts @@ -35,6 +35,7 @@ import { SpServiceRegistration } from '@streampipes/platform-services'; import { ConfigurationService } from '../../shared/configuration.service'; import { DialogService, PanelType } from '@streampipes/shared-ui'; import { SpExtensionsServiceDetailsDialogComponent } from '../../dialog/extensions-service-details/extensions-service-details-dialog.component'; +import { SpExtensionsServiceRunningInstancesDialogComponent } from '../../dialog/extensions-service-running-instances/extensions-service-running-instances-dialog.component'; import { TranslatePipe, TranslateService } from '@ngx-translate/core'; import { FlexDirective, @@ -92,7 +93,7 @@ export class SpRegisteredExtensionsServiceComponent { openServiceDetails(serviceReg: SpServiceRegistration) { this.dialogService.open(SpExtensionsServiceDetailsDialogComponent, { - panelType: PanelType.STANDARD_PANEL, + panelType: PanelType.SLIDE_IN_PANEL, title: this.translateService.instant('Service details'), width: '70vw', data: { @@ -100,4 +101,18 @@ export class SpRegisteredExtensionsServiceComponent { }, }); } + + openRunningInstances(serviceReg: SpServiceRegistration) { + this.dialogService.open( + SpExtensionsServiceRunningInstancesDialogComponent, + { + panelType: PanelType.SLIDE_IN_PANEL, + title: this.translateService.instant('Running instances'), + width: '80vw', + data: { + serviceReg, + }, + }, + ); + } } diff --git a/ui/src/app/configuration/shared/configuration.service.ts b/ui/src/app/configuration/shared/configuration.service.ts index 474d5d8a30..8bee9123c2 100644 --- a/ui/src/app/configuration/shared/configuration.service.ts +++ b/ui/src/app/configuration/shared/configuration.service.ts @@ -27,6 +27,30 @@ import { SpServiceRegistration, } from '@streampipes/platform-services'; +export interface RunningExtensionInstances { + serviceId: string; + adapters: RunningAdapterInstance[]; + pipelineElements: RunningPipelineElementInstance[]; +} + +export interface RunningAdapterInstance { + instanceId: string; + name?: string; + appId?: string; + state: 'RUNNING' | 'STARTING' | 'STOPPING'; + orphaned: boolean; +} + +export interface RunningPipelineElementInstance { + instanceId: string; + name?: string; + appId?: string; + pipelineId?: string; + pipelineName?: string; + type: 'PROCESSOR' | 'SINK' | 'UNKNOWN'; + orphaned: boolean; +} + @Injectable({ providedIn: 'root' }) export class ConfigurationService { private http = inject(HttpClient); @@ -68,6 +92,48 @@ export class ConfigurationService { ); } + getRunningExtensionInstances( + serviceId: string, + ): Observable<RunningExtensionInstances> { + return this.http + .get( + this.getServerUrl() + + `/api/v2/extensions-services/${encodeURIComponent(serviceId)}/running-instances`, + ) + .pipe( + map(response => { + return response as RunningExtensionInstances; + }), + ); + } + + removeAllRunningExtensionInstances(serviceId: string): Observable<object> { + return this.http.delete( + this.getServerUrl() + + `/api/v2/extensions-services/${encodeURIComponent(serviceId)}/running-instances`, + ); + } + + removeRunningAdapterInstance( + serviceId: string, + instanceId: string, + ): Observable<object> { + return this.http.delete( + this.getServerUrl() + + `/api/v2/extensions-services/${encodeURIComponent(serviceId)}/running-instances/adapters/${encodeURIComponent(instanceId)}`, + ); + } + + removeRunningPipelineElementInstance( + serviceId: string, + instanceId: string, + ): Observable<object> { + return this.http.delete( + this.getServerUrl() + + `/api/v2/extensions-services/${encodeURIComponent(serviceId)}/running-instances/pipeline-elements/${encodeURIComponent(instanceId)}`, + ); + } + getExtensionsServiceConfigs(): Observable<SpServiceConfiguration[]> { return this.http .get(
