This is an automated email from the ASF dual-hosted git repository. wiener pushed a commit to branch edge-extensions in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 889d9a0d963df27325efeab5d9ce569cfa80151c Merge: 0d60558 5599765 Author: Patrick Wiener <[email protected]> AuthorDate: Wed Feb 17 10:55:21 2021 +0100 Merge branch 'dev' into edge-extensions .../all_pipeline_elements_jvm.xml | 10 + NOTICE | 4 +- pom.xml | 59 +- streampipes-backend/pom.xml | 26 +- .../backend/StreamPipesResourceConfig.java | 23 +- .../src/main/resources/openapi.yaml | 34 ++ streampipes-backend/src/main/resources/shiro.ini | 6 +- .../pom.xml | 57 +- .../streampipes/client/StreamPipesClient.java | 155 +++++ .../streampipes/client/StreamPipesCredentials.java | 61 +- .../client/annotation/NotYetImplemented.java | 6 +- .../streampipes/client/api/AbstractClientApi.java | 77 +++ .../org/apache/streampipes/client/api/CRUDApi.java | 14 +- .../streampipes/client/api/DataProcessorApi.java | 83 +++ .../apache/streampipes/client/api/DataSinkApi.java | 75 +++ .../streampipes/client/api/DataStreamApi.java | 99 ++++ .../apache/streampipes/client/api/PipelineApi.java | 108 ++++ .../client/api/PipelineElementTemplateApi.java | 63 ++ .../client/api/SupportsDataProcessorApi.java | 6 +- .../client/api/SupportsDataSinkApi.java | 7 +- .../client/api/SupportsDataStreamApi.java | 6 +- .../client/api/SupportsPipelineApi.java | 6 +- .../api/SupportsPipelineElementTemplateApi.java | 6 +- .../streampipes/client/http/DeleteRequest.java | 51 ++ .../apache/streampipes/client/http/GetRequest.java | 51 ++ .../streampipes/client/http/HttpRequest.java | 115 ++++ .../streampipes/client/http/PostRequest.java | 64 ++ .../http/PostRequestWithPayloadResponse.java | 52 ++ .../http/PostRequestWithoutPayloadResponse.java | 47 ++ .../streampipes/client/http/header/Headers.java | 47 ++ .../streampipes/client/live/EventProcessor.java | 7 +- .../streampipes/client/live/KafkaConfig.java | 29 +- .../client/live/SubscriptionManager.java | 107 ++++ .../client/model/StreamPipesClientConfig.java | 76 +++ .../client/serializer/ListSerializer.java | 42 +- .../client/serializer/ObjectSerializer.java | 21 +- .../streampipes/client/serializer/Serializer.java | 32 +- .../client/util/StreamPipesApiPath.java | 59 ++ streampipes-commons/src/main/resources/logback.xml | 3 +- streampipes-connect-container-master/pom.xml | 7 +- .../master/management/AdapterMasterManagement.java | 61 +- .../master/management/SourcesManagement.java | 22 +- .../container/master/rest/SourcesResource.java | 7 +- .../master/management/SourcesManagementTest.java | 30 +- .../init/AdapterWorkerContainerResourceConfig.java | 1 + .../management/HttpServerAdapterManagement.java | 50 ++ .../worker/rest/HttpServerAdapterResource.java | 29 +- .../elements/SendToBrokerReplayAdapterSink.java | 9 +- .../extensions/ExtensionsResourceConfig.java | 4 +- .../PipelineElementContainerResourceConfig.java | 4 +- .../api/AbstractPipelineElementResource.java | 60 +- .../api/DataSourcePipelineElementResource.java | 136 ----- .../api/DataStreamPipelineElementResource.java | 103 ++++ .../api/InvocablePipelineElementResource.java | 4 + ...Resource.java => PipelineTemplateResource.java} | 2 +- .../container/declarer/DataStreamDeclarer.java | 5 +- .../declarer/SemanticEventProducerDeclarer.java | 1 + .../streampipes/container/html/JSONGenerator.java | 1 + .../container/html/model/Description.java | 9 + .../html/page/EventConsumerWelcomePage.java | 98 ++-- .../html/page/EventProcessingAgentWelcomePage.java | 98 ++-- .../html/page/EventProducerWelcomePage.java | 126 ++-- .../html/page/WelcomePageGeneratorImpl.java | 129 ++-- .../container/init/DeclarersSingleton.java | 44 +- .../html/page/WelcomePageGeneratorImplTest.java | 133 ----- .../pom.xml | 19 +- .../dataexplorer/DataLakeManagementV3.java | 297 ++++++++++ .../dataexplorer}/DataLakeNoUserManagementV3.java | 17 +- .../autoagg/AbstractAutoAggregationQuery.java | 70 +++ .../autoagg/FromNowAutoAggregationQuery.java | 49 ++ .../autoagg/GroupedAutoAggregationQuery.java | 48 ++ .../autoagg/TimeBoundAutoAggregationQuery.java | 49 ++ .../streampipes/dataexplorer/model/Order.java | 18 +- .../param/AggregatedTimeBoundQueryParams.java | 50 ++ .../param/AggregatedTimeUnitQueryParams.java | 51 ++ .../dataexplorer/param/CountQueryParams.java | 5 +- .../GroupedAggregatedTimeBoundQueryParams.java | 46 ++ .../dataexplorer/param/GroupedQueryParams.java | 31 +- .../dataexplorer/param/PagingQueryParams.java | 95 +++ .../dataexplorer/param/QueryParams.java | 18 +- .../dataexplorer/param/TimeBoundQueryParams.java | 29 +- .../dataexplorer/param/TimeUnitQueryParams.java | 27 +- .../dataexplorer/query/DataExplorerQuery.java | 84 +++ .../query/DataExplorerQueryBuilder.java | 61 ++ .../dataexplorer/query/DeleteDataQuery.java | 25 +- .../query/GetAggregatedEventsFromNowQuery.java | 47 ++ .../query/GetAggregatedEventsQuery.java | 45 ++ .../query/GetDateFromSortedTableRecord.java | 69 +++ .../query/GetEventsAutoAggregationQuery.java | 6 +- .../dataexplorer/query/GetEventsFromNowQuery.java | 44 ++ .../dataexplorer/query/GetEventsQuery.java | 37 +- .../query/GetGroupedAggregatedEventsQuery.java | 45 ++ .../dataexplorer/query/GetGroupedEventsQuery.java | 43 ++ .../query/GetHeadersWithTypesQuery.java | 45 +- .../dataexplorer/query/GetMaxPagesQuery.java | 36 +- .../query/GetNumberOfRecordsByTimeUnitQuery.java | 52 ++ .../query/GetNumberOfRecordsQuery.java | 51 ++ .../dataexplorer/query/GetPagingEventsQuery.java | 71 +++ .../query/ParameterizedDataExplorerQuery.java | 8 +- .../dataexplorer/template/QueryTemplates.java | 33 +- .../dataexplorer/utils/DataExplorerUtils.java | 47 ++ .../kafka/config/ConsumerConfigFactory.java | 2 +- streampipes-model-client/pom.xml | 8 +- .../model/client/endpoint/RdfEndpointItem.java | 12 +- .../model/client/user/RawUserApiToken.java | 65 ++ .../apache/streampipes/model/client/user/User.java | 32 + .../model/client/user/UserApiToken.java | 62 ++ streampipes-model/pom.xml | 4 +- .../org/apache/streampipes/model/SpDataStream.java | 12 + .../model/base/AbstractStreamPipesEntity.java | 3 + .../model/base/NamedStreamPipesEntity.java | 11 + .../connect/adapter/AdapterStreamDescription.java | 4 +- .../streampipes/model/datalake}/DataResult.java | 2 +- .../model/datalake}/GroupedDataResult.java | 3 +- .../streampipes/model/datalake}/PageResult.java | 3 +- .../model/graph/DataSourceDescription.java | 1 + .../model/monitoring/ConsumedMessagesInfo.java | 67 +++ .../streampipes/model/monitoring/MessagesInfo.java | 37 +- .../model/monitoring/MessagesMetaInfo.java | 33 +- .../monitoring/PipelineElementMonitoringInfo.java | 86 +++ .../model/monitoring/PipelineMonitoringInfo.java | 67 +++ .../model/monitoring/ProducedMessagesInfo.java | 47 ++ .../model/pipeline/PipelineElementStatus.java | 23 +- .../model/staticproperty/AnyStaticProperty.java | 5 + .../staticproperty/CodeInputStaticProperty.java | 5 + .../staticproperty/CollectionStaticProperty.java | 5 + .../staticproperty/ColorPickerStaticProperty.java | 5 + .../model/staticproperty/DomainStaticProperty.java | 8 +- .../model/staticproperty/FileStaticProperty.java | 5 + .../staticproperty/FreeTextStaticProperty.java | 5 + .../model/staticproperty/MappingPropertyNary.java | 5 + .../model/staticproperty/MappingPropertyUnary.java | 5 + .../staticproperty/MatchingStaticProperty.java | 5 + .../model/staticproperty/OneOfStaticProperty.java | 5 + .../staticproperty/RemoteOneOfStaticProperty.java | 10 +- .../model/staticproperty/SecretStaticProperty.java | 5 + .../model/staticproperty/StaticProperty.java | 2 + .../staticproperty/StaticPropertyAlternative.java | 5 + .../staticproperty/StaticPropertyAlternatives.java | 5 + .../model/staticproperty/StaticPropertyGroup.java | 5 + .../staticproperty/StaticPropertyVisitor.java | 53 ++ .../model/template/PipelineElementTemplate.java | 102 ++++ .../template/PipelineElementTemplateConfig.java | 44 +- .../org/apache/streampipes/model/util/Cloner.java | 6 +- .../manager/execution/http/PipelineExecutor.java | 35 +- .../execution/status/SepMonitoringManager.java | 50 -- .../manager/matching/DataSetGroundingSelector.java | 2 - .../manager/monitoring/job/JobManager.java | 64 -- .../manager/monitoring/job/MonitoringJob.java | 81 --- .../monitoring/job/MonitoringJobExecutor.java | 46 -- .../pipeline/PipelineExecutionStatusCollector.java | 37 ++ .../monitoring/pipeline/TopicInfoCollector.java | 258 ++++++++ .../manager/monitoring/runtime/EventGenerator.java | 72 --- .../monitoring/runtime/PipelineObserver.java | 92 --- .../monitoring/runtime/PipelineStreamReplacer.java | 109 ---- .../monitoring/runtime/RandomDataGenerator.java | 74 --- .../monitoring/runtime/RandomEventGenerator.java | 56 -- .../monitoring/runtime/SchemaGenerator.java | 46 -- .../monitoring/runtime/SepStoppedMonitoring.java | 178 ------ .../SepStoppedMonitoringPipelineBuilder.java | 191 ------ .../monitoring/runtime/SimilarStreamFinder.java | 112 ---- .../monitoring/task/CompareDescriptionTask.java | 23 - .../monitoring/task/GetDescriptionTask.java | 71 --- .../manager/monitoring/task/InvokeRuntimeTask.java | 23 - .../manager/monitoring/task/RemoveRuntimeTask.java | 23 - .../manager/monitoring/task/TaskDefinition.java | 53 -- .../manager/monitoring/task/VerifySchemaTask.java | 23 - .../remote/ContainerProvidedOptionsHandler.java | 9 +- .../manager/setup/CouchDbInstallationStep.java | 5 +- .../template/DataProcessorTemplateHandler.java | 19 +- .../manager/template/DataSinkTemplateHandler.java | 17 +- .../template/PipelineElementTemplateHandler.java | 49 ++ .../template/PipelineElementTemplateVisitor.java | 142 +++++ ...epaVerifier.java => DataProcessorVerifier.java} | 4 +- .../{SecVerifier.java => DataSinkVerifier.java} | 4 +- .../{SepVerifier.java => DataStreamVerifier.java} | 27 +- .../manager/verification/ElementVerifier.java | 24 +- .../verification/extractor/TypeExtractor.java | 43 +- .../streampipes/manager/matching/v2/TestUtils.java | 4 +- .../manager/pipeline/TestSerializer.java | 48 -- .../pom.xml | 28 +- .../apache/streampipes/ps}/DataLakeResourceV3.java | 74 ++- .../ps/PipelineElementTemplateResource.java | 197 +++++++ .../streampipes/rest/shared/api/CRUDResource.java | 14 +- .../shared/impl/AbstractSharedRestInterface.java | 61 ++ streampipes-rest/pom.xml | 11 +- .../streampipes/rest/api/IAssetDashboard.java | 41 -- .../rest/api/IMeasurementUnitResource.java | 26 - .../apache/streampipes/rest/api/INotification.java | 34 -- .../streampipes/rest/api/IOntologyKnowledge.java | 70 --- .../rest/api/IOntologyPipelineElement.java | 37 -- .../org/apache/streampipes/rest/api/IPipeline.java | 47 -- .../streampipes/rest/api/IPipelineCache.java | 30 - .../rest/api/IPipelineElementAsset.java | 29 - .../streampipes/rest/api/IPipelineElementFile.java | 33 -- .../rest/api/IPipelineElementRuntimeInfo.java | 27 - ...plicationLink.java => IPipelineMonitoring.java} | 8 +- .../streampipes/rest/api/IPipelineTemplate.java | 33 -- .../apache/streampipes/rest/api/IRdfEndpoint.java | 34 -- .../org/apache/streampipes/rest/api/ISetup.java | 34 -- .../org/apache/streampipes/rest/api/IVersion.java | 27 - .../streampipes/rest/api/IVirtualSensor.java | 29 - .../streampipes/rest/api/IVisualization.java | 26 - .../streampipes/rest/api/dashboard/IDashboard.java | 36 -- .../rest/api/dashboard/IDashboardWidget.java | 36 -- .../rest/api/dashboard/IVisualizablePipeline.java | 32 - .../rest/api/dataexplorer/IDataExplorerWidget.java | 36 -- .../AbstractAuthGuardedRestResource.java} | 23 +- ...estInterface.java => AbstractRestResource.java} | 40 +- .../streampipes/rest/impl/ApplicationLink.java | 79 --- .../streampipes/rest/impl/AssetDashboard.java | 23 +- .../streampipes/rest/impl/Authentication.java | 24 +- .../apache/streampipes/rest/impl/AutoComplete.java | 4 +- .../streampipes/rest/impl/CategoryResource.java | 8 +- .../apache/streampipes/rest/impl/ConsulConfig.java | 27 +- .../rest/impl/ContainerProvidedOptions.java | 2 +- .../org/apache/streampipes/rest/impl/Couchdb.java | 6 +- .../apache/streampipes/rest/impl/DataStream.java | 2 +- .../apache/streampipes/rest/impl/Deployment.java | 2 +- .../rest/impl/InternalPipelineTemplates.java | 31 +- .../streampipes/rest/impl/LabelResource.java | 9 +- .../rest/impl/MeasurementUnitResource.java | 5 +- .../org/apache/streampipes/rest/impl/Node.java | 2 +- .../apache/streampipes/rest/impl/Notification.java | 16 +- .../streampipes/rest/impl/OntologyContext.java | 23 +- .../streampipes/rest/impl/OntologyKnowledge.java | 45 +- .../rest/impl/OntologyMeasurementUnit.java | 10 +- .../rest/impl/OntologyPipelineElement.java | 21 +- .../streampipes/rest/impl/PipelineCache.java | 17 +- .../streampipes/rest/impl/PipelineCategory.java | 2 +- .../rest/impl/PipelineElementAsset.java | 9 +- .../rest/impl/PipelineElementCategory.java | 15 +- .../streampipes/rest/impl/PipelineElementFile.java | 4 +- .../rest/impl/PipelineElementImport.java | 12 +- .../rest/impl/PipelineElementRuntimeInfo.java | 5 +- ...rovidedOptions.java => PipelineMonitoring.java} | 23 +- .../streampipes/rest/impl/PipelineResource.java | 289 +++++++++ .../streampipes/rest/impl/PipelineTemplate.java | 87 +-- .../rest/impl/PipelineWithUserResource.java | 271 --------- .../apache/streampipes/rest/impl/RdfEndpoint.java | 27 +- .../rest/impl/SemanticEventConsumer.java | 2 +- .../rest/impl/SemanticEventProcessingAgent.java | 2 +- .../rest/impl/SemanticEventProducer.java | 35 +- .../org/apache/streampipes/rest/impl/Setup.java | 24 +- .../streampipes/rest/impl/StreamPipesLogs.java | 147 ----- .../org/apache/streampipes/rest/impl/User.java | 35 +- .../org/apache/streampipes/rest/impl/Version.java | 5 +- .../streampipes/rest/impl/VirtualSensor.java | 5 +- .../streampipes/rest/impl/Visualization.java | 4 +- .../impl/dashboard/AbstractDashboardResource.java | 18 +- .../rest/impl/dashboard/DashboardWidget.java | 10 +- .../rest/impl/dashboard/VisualizablePipeline.java | 9 +- ...shboard.java => DataLakeDashboardResource.java} | 2 +- .../rest/impl/datalake/DataLakeManagementV3.java | 652 --------------------- .../impl/datalake/DataLakeNoUserResourceV3.java | 5 +- .../rest/impl/datalake/DataLakeWidgetResource.java | 10 +- .../rest/impl/nouser/FileServingResource.java | 4 +- .../impl/nouser/PipelineElementImportNoUser.java | 8 +- .../rest/impl/nouser/PipelineNoUserResource.java | 4 +- .../rest/management/PipelineManagement.java | 4 +- .../datalake/DataLakeNoUserManagementV3Test.java | 3 +- .../sdk/builder/StreamRequirementsBuilder.java | 8 + .../streampipes/sdk/helpers/Alternatives.java | 15 + .../streampipes/sdk/helpers/EpRequirements.java | 18 + streampipes-serializers-json/pom.xml | 9 +- .../serializers/json/TestGsonSerializer.java | 23 +- .../serializers/json/TestJacksonSerializer.java | 42 ++ .../utils/PipelineElementTemplateHelpers.java | 52 ++ .../streampipes/storage/api/CRUDStorage.java | 16 +- .../streampipes/storage/api/INoSqlStorage.java | 2 + .../api/IPipelineElementDescriptionStorage.java | 31 +- .../api/IPipelineElementTemplateStorage.java | 10 +- .../storage/couchdb/CouchDbStorageManager.java | 5 + .../impl/PipelineElementTemplateStorageImpl.java | 67 +++ .../storage/couchdb/impl/UserStorage.java | 16 +- .../streampipes/storage/couchdb/utils/Utils.java | 6 + .../rdf4j/impl/PipelineElementInMemoryStorage.java | 114 ++-- .../rdf4j/impl/PipelineElementStorageRequests.java | 58 +- .../storage/rdf4j/sparql/QueryBuilder.java | 25 +- streampipes-user-management/pom.xml | 11 +- .../authentication/StreamPipesRealm.java | 40 +- .../user/management/service/TokenService.java | 61 ++ .../user/management/util/TokenUtil.java | 62 ++ .../user/management/util/TestTokenUtil.java | 57 ++ .../apache/streampipes/vocabulary/StreamPipes.java | 5 +- .../wrapper/siddhi/SiddhiAppConfig.java | 70 +++ .../wrapper/siddhi/SiddhiAppConfigBuilder.java | 61 ++ .../wrapper/siddhi/SiddhiQueryBuilder.java | 96 +++ .../siddhi/definition/SiddhiDefinition.java | 9 +- .../siddhi/definition/StreamDefinition.java | 13 +- .../siddhi/definition/WindowDefinition.java | 13 +- .../wrapper/siddhi/engine/SiddhiEngine.java | 9 +- .../wrapper/siddhi/engine/SiddhiEventEngine.java | 2 +- .../siddhi/engine/SiddhiStatementGenerator.java | 5 +- .../siddhi/engine/StreamPipesSiddhiProcessor.java | 4 +- .../engine/callback/SiddhiDebugCallback.java | 4 + .../callback/SiddhiOutputStreamCallback.java | 39 +- .../callback/SiddhiOutputStreamDebugCallback.java | 20 +- .../engine/generator/SiddhiAppGenerator.java | 38 +- .../generator/SiddhiInvocationConfigGenerator.java | 29 +- .../siddhi/output/SiddhiFirstOutputConfig.java | 12 +- .../siddhi/output/SiddhiListOutputConfig.java | 18 +- .../wrapper/siddhi/output/SiddhiOutputConfig.java | 25 +- .../wrapper/siddhi/output/SiddhiOutputType.java | 6 +- .../wrapper/siddhi/query/FromClause.java | 2 +- .../query/{FromClause.java => GroupByClause.java} | 28 +- .../wrapper/siddhi/query/HavingClause.java | 10 +- .../{FromClause.java => InsertIntoClause.java} | 28 +- .../TimeWindowExpression.java => LimitClause.java} | 20 +- .../query/{FromClause.java => OrderByClause.java} | 28 +- .../wrapper/siddhi/query/SelectClause.java | 2 +- .../wrapper/siddhi/query/SiddhiQuery.java | 126 ++++ .../wrapper/siddhi/query/SiddhiStatement.java | 20 +- .../siddhi/query/expression/Expressions.java | 26 +- .../CountExpression.java} | 20 +- .../orderby/OrderByExpression.java} | 29 +- ...owExpression.java => SortWindowExpression.java} | 22 +- ...ression.java => TimeBatchWindowExpression.java} | 7 +- .../expression/window/TimeWindowExpression.java | 1 + .../wrapper/siddhi/utils/SiddhiUtils.java | 27 +- ui/angular.json | 4 +- ui/deployment/app-routing.module.mst | 8 +- ui/deployment/appng5.module.mst | 6 +- ui/package.json | 6 + ui/src/app/add/add.component.html | 3 +- ui/src/app/add/add.component.ts | 4 +- .../endpoint-item/endpoint-item.component.html | 23 +- .../endpoint-item/endpoint-item.component.ts | 14 +- .../apidocs.component.html} | 9 +- .../apidocs.component.scss} | 2 - .../src/app/apidocs/apidocs.component.ts | 32 +- .../src/app/apidocs/apidocs.module.ts | 38 +- ui/src/app/app-routing.module.ts | 8 +- .../app-transport-monitoring.module.ts | 4 +- .../outgoing/outgoing-view.component.html | 12 +- .../transport-summary.component.html | 14 +- .../data-marketplace/data-marketplace.component.ts | 1 + ui/src/app/connect/services/rest.service.ts | 2 +- .../app/core-model/gen/streampipes-model-client.ts | 95 ++- ui/src/app/core-model/gen/streampipes-model.ts | 143 ++++- ui/src/app/core-ui/core-ui.module.ts | 12 +- .../static-any-input/static-any-input.component.ts | 3 - .../static-free-input.component.html | 1 - .../barchart/barchart-widget.component.html} | 20 +- .../barchart/barchart-widget.component.scss} | 4 +- .../widget/barchart/barchart-widget.component.ts | 41 +- .../widget/status/status-widget.component.css} | 0 .../widget/status/status-widget.component.html} | 7 +- .../widget/status/status-widget.component.ts} | 13 +- .../core/components/toolbar/toolbar.component.html | 10 +- .../core/components/toolbar/toolbar.component.scss | 13 +- .../core/components/toolbar/toolbar.component.ts | 11 +- .../components/grid/dashboard-grid.component.html | 6 +- .../components/grid/dashboard-grid.component.ts | 12 +- .../overview/dashboard-overview.component.css | 15 +- .../overview/dashboard-overview.component.html | 19 +- .../components/panel/dashboard-panel.component.css | 13 + .../panel/dashboard-panel.component.html | 39 +- .../components/panel/dashboard-panel.component.ts | 4 +- .../widget/dashboard-widget.component.html | 28 +- .../widget/dashboard-widget.component.ts | 3 +- .../components/widgets/bar-race/bar-race-config.ts | 45 ++ .../bar-race/bar-race-widget.component.html} | 16 +- .../bar-race-widget.component.scss} | 51 +- .../widgets/bar-race/bar-race-widget.component.ts | 165 ++++++ .../components/widgets/base/base-echarts-widget.ts | 59 ++ .../widgets/number/number-widget.component.css | 4 +- .../stacked-line-chart-config.ts | 45 ++ .../stacked-line-chart-widget.component.html} | 16 +- .../stacked-line-chart-widget.component.scss} | 51 +- .../stacked-line-chart-widget.component.ts | 122 ++++ .../components/widgets/status/status-config.ts | 43 ++ .../widgets/status/status-widget.component.html} | 16 +- .../status-widget.component.scss} | 51 +- .../widgets/status/status-widget.component.ts | 67 +++ .../widgets/wordcloud/wordcloud-config.ts | 48 ++ .../wordcloud/wordcloud-widget.component.html} | 17 +- .../wordcloud/wordcloud-widget.component.scss} | 37 +- .../wordcloud/wordcloud-widget.component.ts | 139 +++++ ui/src/app/dashboard/dashboard.module.ts | 17 + ui/src/app/dashboard/registry/widget-registry.ts | 37 +- ui/src/app/dashboard/sdk/ep-requirements.ts | 120 ++-- .../dashboard/sdk/matching/list-property-match.ts | 24 +- .../app/dashboard/sdk/matching/property-match.ts | 8 +- ...pipeline-element-template-config.component.html | 60 ++ ...ipeline-element-template-config.component.scss} | 15 + .../pipeline-element-template-config.component.ts | 79 +++ .../pipeline-element-template-generator.ts} | 33 +- .../dialog/customize/customize.component.html | 76 ++- .../dialog/customize/customize.component.scss | 11 + .../editor/dialog/customize/customize.component.ts | 96 ++- ui/src/app/editor/editor.component.ts | 8 +- ui/src/app/editor/editor.module.ts | 4 +- ui/src/app/home/components/status.component.ts | 15 +- .../login/components/login/login.component.html | 1 + ui/src/app/login/login.module.ts | 2 + .../actions/pipeline-actions.component.ts | 4 +- .../elements/pipeline-elements-row.component.html | 4 +- .../elements/pipeline-elements-row.component.ts | 3 + .../components/model/pipeline-details.model.ts} | 6 +- .../monitoring/pipeline-monitoring.component.html | 69 +++ .../monitoring/pipeline-monitoring.component.scss | 38 +- .../monitoring/pipeline-monitoring.component.ts | 112 ++++ .../pipeline-element-statistics.component.html | 64 ++ .../pipeline-element-statistics.component.scss | 9 +- .../pipeline-element-statistics.component.ts | 140 +++++ .../preview/pipeline-preview.component.html | 2 +- .../pipeline-details.component.html | 7 +- .../pipeline-details/pipeline-details.module.ts | 6 + .../pipeline-overview.component.html | 4 +- .../pipeline-overview.component.ts | 4 +- .../services/pipeline-operations.service.ts | 16 +- .../apis/pipeline-element-template.service.ts | 69 +++ .../apis/pipeline-element.service.ts | 20 +- .../apis/pipeline-monitoring.service.ts | 38 +- ui/src/app/platform-services/platform.module.ts | 6 +- .../profile/components/basic-profile-settings.ts | 58 ++ .../general-profile-settings.component.html | 51 ++ .../general-profile-settings.component.scss} | 20 +- .../general/general-profile-settings.component.ts | 29 +- .../token/token-management-settings.component.html | 90 +++ .../token-management-settings.component.scss} | 65 +- .../token/token-management-settings.component.ts | 70 +++ ui/src/app/profile/profile.component.html | 42 ++ .../profile.component.scss} | 4 +- .../src/app/profile/profile.component.ts | 23 +- .../profile.module.ts} | 45 +- ui/src/app/profile/profile.service.ts | 53 ++ ui/src/scss/main.scss | 1 + ui/src/scss/sp/colors.scss | 4 + ui/src/scss/sp/main.scss | 29 +- ui/src/scss/sp/sp-dialog.scss | 2 +- 432 files changed, 10125 insertions(+), 5973 deletions(-) diff --cc pom.xml index 28b9c5c,3dff2cc..c456af2 --- a/pom.xml +++ b/pom.xml @@@ -57,12 -57,10 +57,12 @@@ <guava.version>27.1-jre</guava.version> <hibernate-validator.version>6.1.5.Final</hibernate-validator.version> <httpclient.version>4.5.13</httpclient.version> + <hawtbuf.version>1.11</hawtbuf.version> + <httpclient.version>4.5.10</httpclient.version> <httpcore.version>4.4.9</httpcore.version> <httpcore-osgi.version>4.4.9</httpcore-osgi.version> - <influxdb.version>2.16</influxdb.version> - <jackson.version>2.10.0</jackson.version> + <influxdb.version>2.21</influxdb.version> + <jackson.version>2.12.1</jackson.version> <jakarta-annotation.version>1.3.5</jakarta-annotation.version> <jakarta-validation.version>2.0.2</jakarta-validation.version> <javapoet.version>1.6.1</javapoet.version> @@@ -871,9 -891,13 +898,14 @@@ <module>streampipes-config</module> <module>streampipes-connect</module> <module>streampipes-connect-container</module> + <module>streampipes-connect-container-master</module> + <module>streampipes-connect-container-worker</module> <module>streampipes-container</module> + <module>streampipes-container-base</module> <module>streampipes-container-embedded</module> + <module>streampipes-container-extensions</module> <module>streampipes-container-standalone</module> ++ <module>streampipes-data-explorer</module> <module>streampipes-dataformat</module> <module>streampipes-dataformat-cbor</module> <module>streampipes-dataformat-fst</module> @@@ -884,9 -909,12 +917,13 @@@ <module>streampipes-messaging-kafka</module> <module>streampipes-messaging</module> <module>streampipes-messaging-jms</module> + <module>streampipes-messaging-mqtt</module> <module>streampipes-model</module> <module>streampipes-model-client</module> + <module>streampipes-model-shared</module> ++ <module>streampipes-node-controller-container</module> <module>streampipes-performance-tests</module> + <module>streampipes-platform-services</module> <module>streampipes-pipeline-management</module> <module>streampipes-sources</module> <module>streampipes-rest</module> @@@ -907,16 -937,7 +946,6 @@@ <module>streampipes-wrapper-siddhi</module> <module>streampipes-wrapper-spark</module> <module>streampipes-wrapper-standalone</module> - <module>streampipes-connect-container-master</module> - <module>streampipes-connect-container-worker</module> - <module>streampipes-container-base</module> - <module>streampipes-node-controller-container</module> - <module>streampipes-messaging-mqtt</module> - <module>streampipes-maven-plugin</module> - <module>streampipes-model-shared</module> - <module>streampipes-container-extensions</module> - <module>streampipes-serializers-jsonld</module> - <module>streampipes-serializers-json</module> - <module>streampipes-data-explorer</module> </modules> <profiles> diff --cc streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/SourcesManagement.java index eb7fba4,0ed7ab6..f2535f7 --- a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/SourcesManagement.java +++ b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/SourcesManagement.java @@@ -179,21 -179,13 +179,17 @@@ public class SourcesManagement String url = adapterDescription.getUri(); ds.setName(adapterDescription.getName()); - ds.setDescription("Description"); - - ds.setUri(url + "/streams"); + ds.setDescription(adapterDescription.getDescription()); + ds.setCorrespondingAdapterId(adapterDescription.getAdapterId()); + ds.setInternallyManaged(true); +// ds.setDeploymentTargetNodeId(adapterDescription.getDeploymentTargetNodeId()); +// ds.setDeploymentTargetNodeHostname(adapterDescription.getDeploymentTargetNodeHostname()); +// ds.setDeploymentTargetNodePort(adapterDescription.getDeploymentTargetNodePort()); + - DataSourceDescription dataSourceDescription = new DataSourceDescription( - url, "Adapter Data Source", - "This data source contains one data stream from the adapters"); - - dataSourceDescription.addEventStream(ds); + ds.setUri(url); - return dataSourceDescription; + return ds; } public void setConnectHost(String connectHost) { diff --cc streampipes-container/src/main/java/org/apache/streampipes/container/declarer/DataStreamDeclarer.java index e3e24c2,9993d4e..09a40cb --- a/streampipes-container/src/main/java/org/apache/streampipes/container/declarer/DataStreamDeclarer.java +++ b/streampipes-container/src/main/java/org/apache/streampipes/container/declarer/DataStreamDeclarer.java @@@ -17,13 -17,11 +17,12 @@@ */ package org.apache.streampipes.container.declarer; + import org.apache.streampipes.model.SpDataStream; - import org.apache.streampipes.model.graph.DataSourceDescription; - public interface DataStreamDeclarer { + public interface DataStreamDeclarer extends Declarer<SpDataStream> { - SpDataStream declareModel(DataSourceDescription sep); + SpDataStream declareModel(); void executeStream(); diff --cc streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java index fa55acc,4befd89..ab31084 --- a/streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java +++ b/streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java @@@ -214,29 -205,9 +206,25 @@@ public class DeclarersSingleton return Http + hostName + Colon + port + route; } + private void checkAndStartExecutableStreams(DataStreamDeclarer declarer) { + if (declarer.isExecutable()) { + declarer.executeStream(); + } + } ++ + public int getPort() { + return port; + } + + public String getHostname() { + return hostName; + } + - private void checkAndStartExecutableStreams(SemanticEventProducerDeclarer sourceDeclarer) { - sourceDeclarer.getEventStreams() - .stream() - .filter(DataStreamDeclarer::isExecutable) - .forEach(es -> { - es.declareModel(sourceDeclarer.declareModel()); - es.executeStream(); - }); - } - + public String getServiceName() { + return serviceName; + } + + public void setServiceName(String serviceName) { + this.serviceName = serviceName; + } } diff --cc streampipes-model/src/main/java/org/apache/streampipes/model/SpDataStream.java index bdc2d70,5e3b483..000458c --- a/streampipes-model/src/main/java/org/apache/streampipes/model/SpDataStream.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/SpDataStream.java @@@ -79,17 -79,11 +79,20 @@@ public class SpDataStream extends Named @RdfProperty(StreamPipes.INDEX) private int index; + @RdfProperty(StreamPipes.HAS_CORRESPONDING_ADAPTER_ID) + private String correspondingAdapterId; + protected List<String> category; + @RdfProperty(StreamPipes.DEPLOYMENT_TARGET_NODE_ID) + private String deploymentTargetNodeId; + + @RdfProperty(StreamPipes.DEPLOYMENT_TARGET_NODE_HOSTNAME) + private String deploymentTargetNodeHostname; + + @RdfProperty(StreamPipes.DEPLOYMENT_TARGET_NODE_PORT) + private Integer deploymentTargetNodePort; + public SpDataStream(String uri, String name, String description, String iconUrl, List<EventStreamQualityDefinition> hasEventStreamQualities, EventGrounding eventGrounding, EventSchema eventSchema) { @@@ -203,27 -195,11 +207,35 @@@ this.index = index; } + public String getDeploymentTargetNodeId() { + return deploymentTargetNodeId; + } + + public void setDeploymentTargetNodeId(String deploymentTargetNodeId) { + this.deploymentTargetNodeId = deploymentTargetNodeId; + } + + public String getDeploymentTargetNodeHostname() { + return deploymentTargetNodeHostname; + } + + public void setDeploymentTargetNodeHostname(String deploymentTargetNodeHostname) { + this.deploymentTargetNodeHostname = deploymentTargetNodeHostname; + } + + public Integer getDeploymentTargetNodePort() { + return deploymentTargetNodePort; + } + + public void setDeploymentTargetNodePort(Integer deploymentTargetNodePort) { + this.deploymentTargetNodePort = deploymentTargetNodePort; + } ++ + public String getCorrespondingAdapterId() { + return correspondingAdapterId; + } + + public void setCorrespondingAdapterId(String correspondingAdapterId) { + this.correspondingAdapterId = correspondingAdapterId; + } } diff --cc streampipes-model/src/main/java/org/apache/streampipes/model/util/Cloner.java index 26c6642,680d5bc..2ac64de --- a/streampipes-model/src/main/java/org/apache/streampipes/model/util/Cloner.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/util/Cloner.java @@@ -188,13 -184,9 +188,13 @@@ public class Cloner } public List<SpDataStream> streams(List<SpDataStream> spDataStreams) { - return spDataStreams.stream().map(s -> new SpDataStream(s)).collect(Collectors.toList()); + return spDataStreams.stream().map(this::mapSequence).collect(Collectors.toList()); } + public List<SpDataStreamRelay> relays(List<SpDataStreamRelay> spDataStreamRelays) { + return spDataStreamRelays.stream().map(s -> new SpDataStreamRelay(s)).collect(Collectors.toList()); + } + public SpDataStream mapSequence(SpDataStream seq) { if (seq instanceof SpDataSet) { return new SpDataSet((SpDataSet) seq); diff --cc streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineExecutor.java index 122156d,c283bad..35d7fd9 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineExecutor.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineExecutor.java @@@ -18,20 -18,7 +18,18 @@@ package org.apache.streampipes.manager.execution.http; +import org.apache.streampipes.manager.data.PipelineGraph; +import org.apache.streampipes.manager.data.PipelineGraphBuilder; +import org.apache.streampipes.manager.data.PipelineGraphHelpers; +import org.apache.streampipes.manager.matching.InvocationGraphBuilder; +import org.apache.streampipes.model.SpDataStream; +import org.apache.streampipes.model.SpDataStreamRelay; +import org.apache.streampipes.model.SpDataStreamRelayContainer; +import org.apache.streampipes.model.base.NamedStreamPipesEntity; +import org.apache.streampipes.model.grounding.EventGrounding; +import org.apache.streampipes.model.pipeline.PipelineElementStatus; +import org.apache.streampipes.sdk.helpers.Tuple2; - import org.lightcouch.DocumentConflictException; import org.apache.streampipes.manager.execution.status.PipelineStatusManager; - import org.apache.streampipes.manager.execution.status.SepMonitoringManager; import org.apache.streampipes.manager.util.TemporaryGraphStorage; import org.apache.streampipes.model.SpDataSet; import org.apache.streampipes.model.base.InvocableStreamPipesEntity; @@@ -83,31 -75,18 +87,27 @@@ public class PipelineExecutor List<InvocableStreamPipesEntity> decryptedGraphs = decryptSecrets(graphs); - graphs.forEach(g -> g.setStreamRequirements(Arrays.asList())); + graphs.forEach(g -> g.setStreamRequirements(Collections.emptyList())); - PipelineOperationStatus status = new GraphSubmitter(pipeline.getPipelineId(), - pipeline.getName(), decryptedGraphs, dataSets) - .invokeGraphs(); + List<SpDataStreamRelayContainer> dataStreamRelayContainers = generateDataStreamRelays(decryptedGraphs); + + PipelineOperationStatus status = new GraphSubmitter( + pipeline.getPipelineId(), + pipeline.getName(), + decryptedGraphs, + dataSets, + dataStreamRelayContainers).invokeGraphs(); if (status.isSuccess()) { storeInvocationGraphs(pipeline.getPipelineId(), graphs, dataSets); - PipelineStatusManager.addPipelineStatus(pipeline.getPipelineId(), - new PipelineStatusMessage(pipeline.getPipelineId(), System.currentTimeMillis(), PipelineStatusMessageType.PIPELINE_STARTED.title(), PipelineStatusMessageType.PIPELINE_STARTED.description())); + PipelineStatusManager.addPipelineStatus( + pipeline.getPipelineId(), + new PipelineStatusMessage(pipeline.getPipelineId(), + System.currentTimeMillis(), + PipelineStatusMessageType.PIPELINE_STARTED.title(), + PipelineStatusMessageType.PIPELINE_STARTED.description())); - if (monitor) { - SepMonitoringManager.addObserver(pipeline.getPipelineId()); - } - if (storeStatus) { setPipelineStarted(pipeline); } @@@ -115,332 -94,15 +115,337 @@@ return status; } + public PipelineOperationStatus stopPipeline() { + List<InvocableStreamPipesEntity> graphs = TemporaryGraphStorage.graphStorage.get(pipeline.getPipelineId()); + List<SpDataSet> dataSets = TemporaryGraphStorage.datasetStorage.get(pipeline.getPipelineId()); + + List<SpDataStreamRelayContainer> dataStreamRelayContainers = generateDataStreamRelays(graphs); + + PipelineOperationStatus status = new GraphSubmitter( + pipeline.getPipelineId(), + pipeline.getName(), + graphs, + dataSets, + dataStreamRelayContainers).detachGraphs(); + + if (status.isSuccess()) { + if (visualize) { + StorageDispatcher + .INSTANCE + .getNoSqlStore() + .getVisualizationStorageApi() + .deleteVisualization(pipeline.getPipelineId()); + } + if (storeStatus) { + setPipelineStopped(pipeline); + } + + PipelineStatusManager.addPipelineStatus(pipeline.getPipelineId(), + new PipelineStatusMessage(pipeline.getPipelineId(), + System.currentTimeMillis(), + PipelineStatusMessageType.PIPELINE_STOPPED.title(), + PipelineStatusMessageType.PIPELINE_STOPPED.description())); + - if (monitor) { - SepMonitoringManager.removeObserver(pipeline.getPipelineId()); - } - + } + return status; + } + + public PipelineOperationStatus updatePipelineDeploymentPartial(Pipeline pipelineOld){ + PipelineOperationStatus status = initPipelineOperationStatus(pipeline.getPipelineId(), pipeline.getName()); + //Adjust Relays in Description using existing configure Method in InvocationGraphBuilder + pipeline.getSepas().forEach(s -> s.setOutputStreamRelays(new ArrayList<>())); + + PipelineGraph oldPipelineGraph = new PipelineGraphBuilder(pipelineOld).buildGraph(); + PipelineGraph newPipelineGraph = new PipelineGraphBuilder(pipeline).buildGraph(); + new InvocationGraphBuilder(newPipelineGraph, pipeline.getPipelineId()).buildGraphs(); + + getDelta(this.pipeline, pipelineOld).forEach(t -> { + List<NamedStreamPipesEntity> predecessors = new ArrayList<>(); + + PipelineGraphHelpers + .findStreams(newPipelineGraph) + .forEach(source -> predecessors.addAll(getPredecessors(source, t.a, newPipelineGraph, new ArrayList<>()))); + List<Tuple2<NamedStreamPipesEntity, NamedStreamPipesEntity>> matchingPredecessors = new ArrayList<>(); + for(NamedStreamPipesEntity e : predecessors){ + matchingPredecessors.add(new Tuple2<>(e, findMatching(e, oldPipelineGraph))); + } + List<NamedStreamPipesEntity> predecessorsOld = new ArrayList<>(); + for(Tuple2<NamedStreamPipesEntity, NamedStreamPipesEntity> pred : matchingPredecessors){ + predecessorsOld.add(pred.b); + } + + PipelineOperationStatus statusStartTarget = startPipelineElement(Collections.singletonList(t.a)); + statusStartTarget.getElementStatus().forEach(status::addPipelineElementStatus); + + //Stop relay to origin + PipelineOperationStatus statusStopRelays = stopRelays(predecessorsOld, t.b); + statusStopRelays.getElementStatus().forEach(status::addPipelineElementStatus); + //start Relay to target + PipelineOperationStatus statusStartRelays = startRelays(predecessors, t.a); + statusStartRelays.getElementStatus().forEach(status::addPipelineElementStatus); + //Stop origin and associated relay + PipelineOperationStatus stopOrigin = stopPipelineElements(Collections.singletonList(t.b)); + stopOrigin.getElementStatus().forEach(status::addPipelineElementStatus); + + //Remove if unnecessary + List<InvocableStreamPipesEntity> graphs = new ArrayList<>(); + graphs.addAll(pipeline.getActions()); + graphs.addAll(pipeline.getSepas()); + List<SpDataSet> dataSets = pipeline.getStreams().stream().filter(s -> s instanceof SpDataSet).map(s -> new + SpDataSet((SpDataSet) s)).collect(Collectors.toList()); + storeInvocationGraphs(pipeline.getPipelineId(), graphs, dataSets); + + }); + return verifyPipelineOperationStatus(status, + "Successfully migrated Pipeline Elements in Pipeline " + pipeline.getName(), + "Could not migrate all Pipeline Elements in Pipeline " + pipeline.getName()); + } + + private PipelineOperationStatus stopPipelineElements(List<NamedStreamPipesEntity> graphs){ + List<InvocableStreamPipesEntity> invocations = new ArrayList<>(); + graphs.stream().filter(i -> i instanceof InvocableStreamPipesEntity).forEach(i -> invocations.add((InvocableStreamPipesEntity) i)); + + if (invocations.isEmpty()) { + PipelineOperationStatus ret = initPipelineOperationStatus(pipeline.getPipelineId(), pipeline.getName()); + ret.setSuccess(true); + return ret; + } + List<SpDataStreamRelayContainer> dataStreamRelayContainers = generateDataStreamRelays(invocations); + return new GraphSubmitter(pipeline.getPipelineId(), + pipeline.getName(), invocations, new ArrayList<>(), dataStreamRelayContainers) + .detachGraphs(); + } + + private PipelineOperationStatus startPipelineElement(List<NamedStreamPipesEntity> graphs){ + List<InvocableStreamPipesEntity> invocations = new ArrayList<>(); + graphs.stream().filter(i -> i instanceof InvocableStreamPipesEntity).forEach(i -> invocations.add((InvocableStreamPipesEntity) i)); + + if (invocations.isEmpty()) { + PipelineOperationStatus ret = initPipelineOperationStatus(pipeline.getPipelineId(), pipeline.getName()); + ret.setSuccess(true); + return ret; + } + List<InvocableStreamPipesEntity> decryptedGraphs = decryptSecrets(invocations); + List<SpDataStreamRelayContainer> dataStreamRelayContainers = generateDataStreamRelays(invocations); + return new GraphSubmitter(pipeline.getPipelineId(), + pipeline.getName(), decryptedGraphs, new ArrayList<>(), dataStreamRelayContainers) + .invokeGraphs(); + } + + private PipelineOperationStatus stopRelays(List<NamedStreamPipesEntity> predecessors, InvocableStreamPipesEntity target){ + Map<NamedStreamPipesEntity, SpDataStreamRelayContainer> relays = new HashMap<>(); + + findRelays(relays, predecessors, target); + + if (relays.isEmpty()) { + PipelineOperationStatus ret = initPipelineOperationStatus(pipeline.getPipelineId(), pipeline.getName()); + ret.setSuccess(true); + return ret; + } + return new GraphSubmitter(pipeline.getPipelineId(), + pipeline.getName(), new ArrayList<>(), new ArrayList<>(), new ArrayList<>()).detachRelays(relays); + } + + private PipelineOperationStatus startRelays(List<NamedStreamPipesEntity> predecessors, InvocableStreamPipesEntity target){ + Map<NamedStreamPipesEntity, SpDataStreamRelayContainer> relays = new HashMap<>(); + findRelays(relays, predecessors, target); + + if (relays.isEmpty()) { + PipelineOperationStatus ret = initPipelineOperationStatus(pipeline.getPipelineId(), pipeline.getName()); + ret.setSuccess(true); + return ret; + } + return new GraphSubmitter(pipeline.getPipelineId(), + pipeline.getName(), new ArrayList<>(), new ArrayList<>(), new ArrayList<>()).invokeRelays(relays); + } + + private void findRelays(Map<NamedStreamPipesEntity, SpDataStreamRelayContainer> relays, + List<NamedStreamPipesEntity> predecessors, InvocableStreamPipesEntity target){ + + predecessors.forEach(pred -> { + List<SpDataStreamRelay> dataStreamRelays = new ArrayList<>(); + SpDataStreamRelayContainer dsRelayContainer = new SpDataStreamRelayContainer(); + + if (pred instanceof DataProcessorInvocation){ + //Data Processor + DataProcessorInvocation processorInvocation = (DataProcessorInvocation) pred; + dataStreamRelays.addAll(processorInvocation.getOutputStreamRelays().stream(). + filter(r -> + target.getInputStreams().stream().map(s -> + s.getEventGrounding().getTransportProtocol().getTopicDefinition().getActualTopicName()) + .collect(Collectors.toSet()).contains(r.getEventGrounding().getTransportProtocol() + .getTopicDefinition().getActualTopicName())) + .collect(Collectors.toList())); + dsRelayContainer.setRunningStreamRelayInstanceId(pipeline.getPipelineId()); + dsRelayContainer.setEventRelayStrategy(pipeline.getEventRelayStrategy()); + dsRelayContainer.setName(processorInvocation.getName() + " (Stream Relay)"); + dsRelayContainer.setInputGrounding(new EventGrounding(processorInvocation.getOutputStream().getEventGrounding())); + dsRelayContainer.setDeploymentTargetNodeId(processorInvocation.getDeploymentTargetNodeId()); + dsRelayContainer.setDeploymentTargetNodeHostname(processorInvocation.getDeploymentTargetNodeHostname()); + dsRelayContainer.setDeploymentTargetNodePort(processorInvocation.getDeploymentTargetNodePort()); + dsRelayContainer.setOutputStreamRelays(dataStreamRelays); + relays.put(pred, dsRelayContainer); + } else if (pred instanceof SpDataStream){ + //DataStream + SpDataStream dataStream = (SpDataStream) pred; + if (!matchingDeploymentTargets(dataStream, target)){ + //There is a relay that needs to be removed + dataStreamRelays.add(new SpDataStreamRelay(new EventGrounding(target.getInputStreams() + .get(getIndex(pred.getDOM(), target)) + .getEventGrounding()))); + + dsRelayContainer.setRunningStreamRelayInstanceId(pipeline.getPipelineId()); + dsRelayContainer.setEventRelayStrategy(pipeline.getEventRelayStrategy()); + dsRelayContainer.setName(dataStream.getName() + " (Stream Relay)"); + dsRelayContainer.setInputGrounding(new EventGrounding(dataStream.getEventGrounding())); + dsRelayContainer.setDeploymentTargetNodeId(dataStream.getDeploymentTargetNodeId()); + dsRelayContainer.setDeploymentTargetNodeHostname(dataStream.getDeploymentTargetNodeHostname()); + dsRelayContainer.setDeploymentTargetNodePort(dataStream.getDeploymentTargetNodePort()); + dsRelayContainer.setOutputStreamRelays(dataStreamRelays); + relays.put(pred, dsRelayContainer); + } + } + }); + + } + + private List<Tuple2<DataProcessorInvocation, DataProcessorInvocation>> getDelta(Pipeline pipelineX, Pipeline pipelineY){ + List<Tuple2<DataProcessorInvocation, DataProcessorInvocation>> delta = new ArrayList<>(); + pipelineX.getSepas().forEach(iX -> { + if (pipelineY.getSepas().stream().filter(iY -> iY.getElementId().equals(iX.getElementId())). + filter(iY -> iY.getDeploymentTargetNodeId().equals(iX.getDeploymentTargetNodeId())).count() == 0){ + Optional<DataProcessorInvocation> invocationY = pipelineY.getSepas().stream(). + filter(iY -> iY.getDeploymentRunningInstanceId().equals(iX.getDeploymentRunningInstanceId())).findFirst(); + invocationY.ifPresent(dataProcessorInvocation -> delta.add(new Tuple2<>(iX, dataProcessorInvocation))); + } + }); + return delta; + } + + private List<NamedStreamPipesEntity> getPredecessors(NamedStreamPipesEntity source, + InvocableStreamPipesEntity target, PipelineGraph pipelineGraph, + List<NamedStreamPipesEntity> foundPredecessors){ + //TODO: Check if this works for all graph topologies + if (pipelineGraph.outgoingEdgesOf(source) + .stream() + .map(pipelineGraph::getEdgeTarget) + .map(g -> (InvocableStreamPipesEntity) g) + .collect(Collectors.toSet()).contains(target)){ + foundPredecessors.add(source); + } else{ + List<NamedStreamPipesEntity> successors = pipelineGraph.outgoingEdgesOf(source) + .stream() + .map(pipelineGraph::getEdgeTarget) + .map(g -> (InvocableStreamPipesEntity) g) + .collect(Collectors.toList()); + if (successors.isEmpty()) return foundPredecessors; + successors.forEach(successor -> getPredecessors(successor, target, pipelineGraph, foundPredecessors)); + } + return foundPredecessors; + } + + private NamedStreamPipesEntity findMatching(NamedStreamPipesEntity entity, PipelineGraph pipelineGraph){ + AtomicReference<NamedStreamPipesEntity> match = new AtomicReference<>(); + PipelineGraphHelpers.findStreams(pipelineGraph).forEach(ds -> { + NamedStreamPipesEntity foundEntity = compareGraphs(ds, entity, pipelineGraph, new ArrayList<>()); + if (foundEntity != null) match.set(foundEntity); + }); + return match.get(); + } + + private NamedStreamPipesEntity compareGraphs(NamedStreamPipesEntity source, NamedStreamPipesEntity searchedEntity, PipelineGraph pipelineGraph, List<NamedStreamPipesEntity> successors){ + if(source.getDOM().equals(searchedEntity.getDOM())) return source; + else if (successors.isEmpty()) + successors = pipelineGraph.outgoingEdgesOf(source) + .stream() + .map(pipelineGraph::getEdgeTarget) + .map(g -> (InvocableStreamPipesEntity) g) + .collect(Collectors.toList()); + Optional<NamedStreamPipesEntity> successor= successors.stream().findFirst(); + if (successor.isPresent()) { + successors.remove(successor.get()); + return compareGraphs(successor.get(), searchedEntity, pipelineGraph, successors); + } + return null; + } + + private PipelineOperationStatus initPipelineOperationStatus(String pipelineId, String pipelineName) { + //Duplicate from method in GraphSubmitter + PipelineOperationStatus status = new PipelineOperationStatus(); + status.setPipelineId(pipelineId); + status.setPipelineName(pipelineName); + return status; + } + + private PipelineOperationStatus verifyPipelineOperationStatus(PipelineOperationStatus status, String successMessage, + String errorMessage) { + //Duplicate from method in GraphSubmitter + status.setSuccess(status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess)); + if (status.isSuccess()) { + status.setTitle(successMessage); + } else { + status.setTitle(errorMessage); + } + return status; + } + + private String extractTopic(EventGrounding eg) { + return eg.getTransportProtocol().getTopicDefinition().getActualTopicName(); + } + + private List<SpDataStreamRelayContainer> generateDataStreamRelays(List<InvocableStreamPipesEntity> graphs) { + Set<String> topicSet = new HashSet<>(); + List<SpDataStreamRelayContainer> dsRelayContainers = new ArrayList<>(); + SpDataStreamRelayContainer dsRelayContainer = new SpDataStreamRelayContainer(); + List<SpDataStreamRelay> dataStreamRelays = new ArrayList<>(); + + graphs.stream() + .filter(g -> pipeline.getStreams().stream() + .filter(ds -> topicSet.add(extractTopic(ds.getEventGrounding()))) + .anyMatch(ds -> (!matchingDeploymentTargets(ds, g) && connected(ds, g)))) + .forEach(g -> pipeline.getStreams() + .forEach(ds -> { + dsRelayContainer.setRunningStreamRelayInstanceId(pipeline.getPipelineId()); + dsRelayContainer.setEventRelayStrategy(pipeline.getEventRelayStrategy()); + dsRelayContainer.setName(ds.getName() + " (Stream Relay)"); + dsRelayContainer.setInputGrounding(new EventGrounding(ds.getEventGrounding())); + dsRelayContainer.setDeploymentTargetNodeId(ds.getDeploymentTargetNodeId()); + dsRelayContainer.setDeploymentTargetNodeHostname(ds.getDeploymentTargetNodeHostname()); + dsRelayContainer.setDeploymentTargetNodePort(ds.getDeploymentTargetNodePort()); + + dataStreamRelays.add(new SpDataStreamRelay(new EventGrounding(g.getInputStreams() + .get(getIndex(ds.getDOM(), g)) + .getEventGrounding()))); + }) + ); + + dsRelayContainer.setOutputStreamRelays(dataStreamRelays); + dsRelayContainers.add(dsRelayContainer); + + return dsRelayContainers; + } + + private boolean matchingDeploymentTargets(SpDataStream source, InvocableStreamPipesEntity target) { + return source.getDeploymentTargetNodeId().equals(target.getDeploymentTargetNodeId()); + } + + private boolean connected(SpDataStream source, InvocableStreamPipesEntity target) { + int index = getIndex(source.getDOM(), target); + if (index != -1) { + return target.getConnectedTo().get(index).equals(source.getDOM()); + } + return false; + } + + private void updateGroupIds(InvocableStreamPipesEntity entity) { + entity.getInputStreams() + .stream() + .filter(is -> is.getEventGrounding().getTransportProtocol() instanceof KafkaTransportProtocol) + .map(is -> is.getEventGrounding().getTransportProtocol()) + .map(KafkaTransportProtocol.class::cast) + .forEach(tp -> tp.setGroupId(UUID.randomUUID().toString())); + } + private List<InvocableStreamPipesEntity> decryptSecrets(List<InvocableStreamPipesEntity> graphs) { List<InvocableStreamPipesEntity> decryptedGraphs = new ArrayList<>(); graphs.stream().map(g -> { diff --cc streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/ConsulConfig.java index db9e3db,f0327fb..743bcc8 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/ConsulConfig.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/ConsulConfig.java @@@ -18,13 -18,8 +18,9 @@@ package org.apache.streampipes.rest.impl; - import static org.apache.streampipes.container.util.ConsulUtil.updateConfig; - import com.google.gson.Gson; import com.google.gson.JsonObject; +import org.apache.streampipes.sdk.helpers.Tuple2; - import org.slf4j.Logger; - import org.slf4j.LoggerFactory; import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.config.backend.MessagingSettings; import org.apache.streampipes.config.consul.ConsulSpConfig; @@@ -56,10 -48,9 +49,9 @@@ public class ConsulConfig extends Abstr @GET @Produces(MediaType.APPLICATION_JSON) @GsonWithIds - @Override public Response getAllServiceConfigs() { LOG.info("Request for all service configs"); - Map<String, String> peServices = ConsulUtil.getPeServices(); + Map<String, Tuple2<String,String>> peServices = ConsulUtil.getPeServices(); List<PeConfig> peConfigs = new LinkedList<>(); @@@ -180,21 -164,8 +168,20 @@@ return ok(); } - public List<ConfigItem> getConfigForService(String serviceId) { - Map<String, String> keyValues = ConsulUtil.getKeyValue(ConsulSpConfig.SERVICE_ROUTE_PREFIX + serviceId); + public List<ConfigItem> getConfigForService(String serviceId, String tag) { - //Map<String, String> keyValues = ConsulUtil.getKeyValue(ConsulSpConfig.SERVICE_ROUTE_PREFIX + serviceId); + Map<String, String> keyValues = ConsulUtil.getKeyValue( + ConsulSpConfig.SERVICE_ROUTE_PREFIX + + serviceId.substring(serviceId.indexOf("/")+1) + + "/" + + ConsulSpConfig.BASE_PREFIX + + "/" + + tag); + + keyValues.putAll(ConsulUtil.getKeyValue( + ConsulSpConfig.SERVICE_ROUTE_PREFIX + + serviceId.substring(serviceId.indexOf("/")+1) + + "/" + + ConsulSpConfig.CONFIG_PREFIX)); List<ConfigItem> configItems = new LinkedList<>(); diff --cc streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Node.java index df43c35,0000000..d0f8c9f mode 100644,000000..100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Node.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Node.java @@@ -1,105 -1,0 +1,105 @@@ +/* + * 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.rest.impl; + +import org.apache.streampipes.manager.node.NodeClusterManager; +import org.apache.streampipes.model.message.NotificationType; +import org.apache.streampipes.model.message.Notifications; +import org.apache.streampipes.model.node.NodeInfoDescription; +import org.apache.streampipes.rest.api.INode; +import org.apache.streampipes.rest.shared.annotation.JacksonSerialized; + +import javax.ws.rs.*; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; + +@Path("/v2/users/{username}/nodes") - public class Node extends AbstractRestInterface implements INode { ++public class Node extends AbstractRestResource implements INode { + + private static final String ACTIVATE = "activate"; + private static final String DEACTIVATE = "deactivate"; + + @POST + @JacksonSerialized + @Consumes(MediaType.APPLICATION_JSON) + @Produces(MediaType.APPLICATION_JSON) + @Override + public Response addNode(@PathParam("username") String username, NodeInfoDescription desc) { + NodeClusterManager.addNode(desc); + return statusMessage(Notifications.success(NotificationType.STORAGE_SUCCESS)); + } + + @PUT + @JacksonSerialized + @Path("/{nodeControllerId}") + @Consumes(MediaType.APPLICATION_JSON) + @Produces(MediaType.APPLICATION_JSON) + @Override + public Response updateNode(@PathParam("username") String username, + @PathParam("nodeControllerId") String nodeControllerId, NodeInfoDescription desc) { + return statusMessage(NodeClusterManager.updateNode(desc)); + } + + @POST + @JacksonSerialized + @Path("/{action}/{nodeControllerId}") + @Consumes(MediaType.APPLICATION_JSON) + @Produces(MediaType.APPLICATION_JSON) + @Override + public Response changeNodeState(@PathParam("action") String action, + @PathParam("username") String username, + @PathParam("nodeControllerId") String nodeControllerId) { + boolean success = false; + if (action.equals(ACTIVATE)) { + success = NodeClusterManager.activateNode(nodeControllerId); + } else if (action.equals(DEACTIVATE)) { + success = NodeClusterManager.deactivateNode(nodeControllerId); + } + if (success) { + return statusMessage(Notifications.success(NotificationType.OPERATION_SUCCESS)); + } else { + return statusMessage(Notifications.error(NotificationType.NODE_STATE_UPDATE_ERROR)); + } + } + + @DELETE + @Path("/{nodeControllerId}") + @Override + public Response deleteNode(@PathParam("username") String username, + @PathParam("nodeControllerId") String nodeControllerId) { + NodeClusterManager.deleteNode(nodeControllerId); + return statusMessage(Notifications.success(NotificationType.REMOVED_NODE)); + } + + @GET + @Path("/available") + @JacksonSerialized + @Produces(MediaType.APPLICATION_JSON) + @Override + public Response getAvailableNodes() { + return ok(NodeClusterManager.getAvailableNodes()); + } + + @GET + @JacksonSerialized + @Produces(MediaType.APPLICATION_JSON) + @Override + public Response getNodes() { + return ok(NodeClusterManager.getAllNodes()); + } +} diff --cc streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java index 0000000,6c58371..74b71cf mode 000000,100644..100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java @@@ -1,0 -1,267 +1,289 @@@ + /* + * 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.rest.impl; + + import com.google.gson.JsonSyntaxException; + import io.swagger.v3.oas.annotations.Hidden; + import io.swagger.v3.oas.annotations.Operation; + import io.swagger.v3.oas.annotations.media.ArraySchema; + import io.swagger.v3.oas.annotations.media.Content; + import io.swagger.v3.oas.annotations.media.Schema; + import io.swagger.v3.oas.annotations.responses.ApiResponse; + import org.apache.streampipes.commons.exceptions.*; + import org.apache.streampipes.manager.execution.status.PipelineStatusManager; + import org.apache.streampipes.manager.operations.Operations; + import org.apache.streampipes.model.SpDataSet; + import org.apache.streampipes.model.client.exception.InvalidConnectionException; + import org.apache.streampipes.model.message.Notification; + import org.apache.streampipes.model.message.NotificationType; + import org.apache.streampipes.model.message.Notifications; + import org.apache.streampipes.model.message.SuccessMessage; + import org.apache.streampipes.model.pipeline.Pipeline; + import org.apache.streampipes.model.pipeline.PipelineOperationStatus; + import org.apache.streampipes.rest.management.PipelineManagement; + import org.apache.streampipes.rest.shared.annotation.JacksonSerialized; + import org.slf4j.Logger; + import org.slf4j.LoggerFactory; + + import javax.ws.rs.*; + import javax.ws.rs.core.MediaType; + import javax.ws.rs.core.Response; + import java.util.Date; + import java.util.UUID; + + @Path("/v2/users/{username}/pipelines") + public class PipelineResource extends AbstractAuthGuardedRestResource { + + private static final Logger logger = LoggerFactory.getLogger(PipelineResource.class); + + public Response getAvailable(String username) { + // TODO Auto-generated method stub + return null; + } + + public Response getFavorites(String username) { + // TODO Auto-generated method stub + return null; + } + + @GET + @Produces(MediaType.APPLICATION_JSON) + @Path("/own") + @JacksonSerialized + @Operation(summary = "Get all pipelines assigned to the current user", + tags = {"Pipeline"}, + responses = { + @ApiResponse(content = { + @Content( + mediaType = "application/json", + array = @ArraySchema(schema = @Schema(implementation = Pipeline.class))) + })}) + public Response getOwn(@PathParam("username") String username) { + return ok(getUserService() + .getOwnPipelines(username)); + } + + @GET + @Produces(MediaType.APPLICATION_JSON) + @Path("/system") + @JacksonSerialized + @Operation(summary = "Get all system pipelines assigned to the current user", + tags = {"Pipeline"}) + public Response getSystemPipelines() { + return ok(getPipelineStorage().getSystemPipelines()); + } + + public Response addFavorite(String username, String elementUri) { + // TODO Auto-generated method stub + return null; + } + + public Response removeFavorite(String username, String elementUri) { + // TODO Auto-generated method stub + return null; + } + + @GET + @Produces(MediaType.APPLICATION_JSON) + @Path("/{pipelineId}/status") + @JacksonSerialized + @Operation(summary = "Get the pipeline status of a given pipeline", + tags = {"Pipeline"}) + public Response getPipelineStatus(@PathParam("username") String username, @PathParam("pipelineId") String pipelineId) { + return ok(PipelineStatusManager.getPipelineStatus(pipelineId, 5)); + } + + @DELETE + @Path("/{pipelineId}") + @Produces(MediaType.APPLICATION_JSON) + @JacksonSerialized + @Operation(summary = "Delete a pipeline with a given id", + tags = {"Pipeline"}) + public Response removeOwn(@PathParam("username") String username, @PathParam("pipelineId") String elementUri) { + getPipelineStorage().deletePipeline(elementUri); + return statusMessage(Notifications.success("Pipeline deleted")); + } + + @GET + @Path("/{pipelineId}") + @Produces(MediaType.APPLICATION_JSON) + @JacksonSerialized + @Operation(summary = "Get a specific pipeline with the given id", + tags = {"Pipeline"}) + public Response getElement(@PathParam("username") String username, @PathParam("pipelineId") String pipelineId) { + return ok(getPipelineStorage().getPipeline(pipelineId)); + } + + @Path("/{pipelineId}/start") + @GET + @Produces(MediaType.APPLICATION_JSON) + @JacksonSerialized + @Operation(summary = "Start the pipeline with the given id", + tags = {"Pipeline"}) + public Response start(@PathParam("username") String username, @PathParam("pipelineId") String pipelineId) { + try { + Pipeline pipeline = getPipelineStorage() + .getPipeline(pipelineId); + PipelineOperationStatus status = Operations.startPipeline(pipeline); + return ok(status); + } catch (Exception e) { + e.printStackTrace(); + return statusMessage(Notifications.error(NotificationType.UNKNOWN_ERROR)); + } + } + + @Path("/{pipelineId}/stop") + @GET + @Produces(MediaType.APPLICATION_JSON) + @JacksonSerialized + @Operation(summary = "Stop the pipeline with the given id", + tags = {"Pipeline"}) + public Response stop(@PathParam("username") String username, @PathParam("pipelineId") String pipelineId) { + logger.info("User: " + username + " stopped pipeline: " + pipelineId); + PipelineManagement pm = new PipelineManagement(); + return pm.stopPipeline(pipelineId); + } + + @POST + @Produces(MediaType.APPLICATION_JSON) + @JacksonSerialized + @Operation(summary = "Store a new pipeline", + tags = {"Pipeline"}) + public Response addPipeline(@PathParam("username") String username, Pipeline pipeline) { + String pipelineId = UUID.randomUUID().toString(); + pipeline.setPipelineId(pipelineId); + pipeline.setRunning(false); + pipeline.setCreatedByUser(username); + pipeline.setCreatedAt(new Date().getTime()); + pipeline.getSepas().forEach(processor -> processor.setCorrespondingUser(username)); + pipeline.getActions().forEach(action -> action.setCorrespondingUser(username)); + //userService.addOwnPipeline(username, pipeline); + Operations.storePipeline(pipeline); + SuccessMessage message = Notifications.success("Pipeline stored"); + message.addNotification(new Notification("id", pipelineId)); + return ok(message); + } + + @Path("/recommend") + @POST + @Produces(MediaType.APPLICATION_JSON) + @JacksonSerialized + @Hidden + public Response recommend(@PathParam("username") String email, Pipeline pipeline) { + try { + return ok(Operations.findRecommendedElements(email, pipeline)); + } catch (JsonSyntaxException e) { + return constructErrorMessage(new Notification(NotificationType.UNKNOWN_ERROR, + e.getMessage())); + } catch (NoSuitableSepasAvailableException e) { + return constructErrorMessage(new Notification(NotificationType.NO_SEPA_FOUND, + e.getMessage())); + } catch (Exception e) { + e.printStackTrace(); + return constructErrorMessage(new Notification(NotificationType.UNKNOWN_ERROR, + e.getMessage())); + } + } + + @Path("/update/dataset") + @POST + @Produces(MediaType.APPLICATION_JSON) + @Consumes(MediaType.APPLICATION_JSON) + @JacksonSerialized + @Hidden + public Response updateDataSet(SpDataSet spDataSet, @PathParam("username") + String username) { + return ok(Operations.updateDataSet(spDataSet)); + } + + @Path("/update") + @POST + @Produces(MediaType.APPLICATION_JSON) + @Consumes(MediaType.APPLICATION_JSON) + @JacksonSerialized + @Hidden + public Response update(Pipeline pipeline, @PathParam("username") String username) { + try { + return ok(Operations.validatePipeline(pipeline, true, username)); + } catch (JsonSyntaxException e) { + return badRequest(new Notification(NotificationType.UNKNOWN_ERROR, + e.getMessage())); + } catch (NoMatchingSchemaException e) { + return badRequest(new Notification(NotificationType.NO_VALID_CONNECTION, + e.getMessage())); + } catch (NoMatchingFormatException e) { + return badRequest(new Notification(NotificationType.NO_MATCHING_FORMAT_CONNECTION, + e.getMessage())); + } catch (NoMatchingProtocolException e) { + return badRequest(new Notification(NotificationType.NO_MATCHING_PROTOCOL_CONNECTION, + e.getMessage())); + } catch (RemoteServerNotAccessibleException | NoMatchingJsonSchemaException e) { + return serverError(new Notification(NotificationType.REMOTE_SERVER_NOT_ACCESSIBLE + , e.getMessage())); + } catch (InvalidConnectionException e) { + return badRequest(e.getErrorLog()); + } catch (Exception e) { + e.printStackTrace(); + return serverError(new Notification(NotificationType.UNKNOWN_ERROR, + e.getMessage())); + } + } + + @PUT + @Path("/{pipelineId}") + @Produces(MediaType.APPLICATION_JSON) + @JacksonSerialized + @Operation(summary = "Update an existing pipeline", + tags = {"Pipeline"}) + public Response overwritePipeline(@PathParam("username") String username, + @PathParam("pipelineId") String pipelineId, + Pipeline pipeline) { + Pipeline storedPipeline = getPipelineStorage().getPipeline(pipelineId); + storedPipeline.setActions(pipeline.getActions()); + storedPipeline.setSepas(pipeline.getSepas()); + storedPipeline.setActions(pipeline.getActions()); + storedPipeline.setCreatedAt(System.currentTimeMillis()); + storedPipeline.setPipelineCategories(pipeline.getPipelineCategories()); + Operations.updatePipeline(storedPipeline); + return statusMessage(Notifications.success("Pipeline modified")); + } + ++ @POST ++ @Path("/migrate/{pipelineId}") ++ @Produces(MediaType.APPLICATION_JSON) ++ @JacksonSerialized ++ @Operation(summary = "Migrate pipeline elements to new node", ++ tags = {"Pipeline"}) ++ public Response migratePipelineProcessors(@PathParam("username") String username, ++ @PathParam("pipelineId") String pipelineId, ++ Pipeline pipelineNew) { ++ try { ++ Pipeline pipelineOld = getPipelineStorage() ++ .getPipeline(pipelineId); ++ PipelineOperationStatus status = Operations.updatePipelineDeploymentPartial(pipelineOld, pipelineNew, ++ true,true, true); ++ Operations.overwritePipeline(pipelineNew); ++ return ok(status); ++ } catch (Exception e) { ++ e.printStackTrace(); ++ return statusMessage(Notifications.error(NotificationType.UNKNOWN_ERROR)); ++ } ++ } ++ + } diff --cc streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/INoSqlStorage.java index a42f82c,856c64a..06fdfea --- a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/INoSqlStorage.java +++ b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/INoSqlStorage.java @@@ -55,5 -55,6 +55,7 @@@ public interface INoSqlStorage IVisualizablePipelineStorage getVisualizablePipelineStorage(); + IPipelineElementTemplateStorage getPipelineElementTemplateStorage(); + + INodeInfoStorage getNodeStorage(); } diff --cc streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java index a88c8db,fc1d4b8..269fb42 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java @@@ -111,9 -111,9 +111,14 @@@ public enum CouchDbStorageManager imple } @Override + public IPipelineElementTemplateStorage getPipelineElementTemplateStorage() { + return new PipelineElementTemplateStorageImpl(); + } + ++ @Override + public INodeInfoStorage getNodeStorage() { + return new NodeInfoStorageImpl(); + } + } diff --cc streampipes-vocabulary/src/main/java/org/apache/streampipes/vocabulary/StreamPipes.java index e2a2150,d269543..5a78b2a --- a/streampipes-vocabulary/src/main/java/org/apache/streampipes/vocabulary/StreamPipes.java +++ b/streampipes-vocabulary/src/main/java/org/apache/streampipes/vocabulary/StreamPipes.java @@@ -418,75 -402,7 +418,78 @@@ public class StreamPipes public static final String USER_DEFINED_OUTPUT_STRATEGY = NS + "UserDefinedOutputStrategy"; public static final String PE_CONFIGURED = NS + "isPeConfigured" ; - public static final String HAS_REQUIRED_FILETYPES = NS + "hasRequiredFiletypes" ; + public static final String HAS_EVENT_RELAY = NS + "hasEventRelay"; + public static final String HAS_EVENT_RELAY_STRATEGY = NS + "hasEventRelayStrategy"; + public static final String DATA_STREAM_RELAY_CONTAINER = NS + "DataStreamRelayContainer" ; + public static final String DATA_STREAM_RELAY_RUNNING_INSTANCE_ID = NS + "hasDataStreamRelayRunningInstanceId"; + + public static final String STATIC_NODE_METADATA = NS + "StaticNodeMetadata"; + public static final String HAS_GEO_LOCATION = NS + "hasGeoLocation"; + public static final String HAS_NODE_TAGS = NS + "hasNodeTags"; + public static final String NODE_TYPE = NS + "isNodeType"; + public static final String NODE_MODEL = NS + "isNodeModel"; + public static final String GEO_LOCATION = NS + "geoLocation"; + public static final String GEO_LOCATION_LATITUDE = NS + "hasLatitude"; + public static final String GEO_LOCATION_LONGITUDE = NS + "hasLongitude"; + public static final String NODE_INFO_DESCRIPTION = NS + "NodeInfoDescription"; + public static final String HAS_STATIC_NODE_METADATA = NS + "hasStaticNodeMetadata"; + public static final String HAS_NODE_RESOURCES = NS + "hasNodeResources"; + + public static final String HAS_CONTAINER = NS + "hasContainer"; + public static final String DEPLOYMENT_CONTAINER = NS + "DeploymentContainer"; + public static final String NODE_RESOURCE = NS + "NodeResource"; + public static final String NODE_HARDWARE_RESOURCE = NS + "NodeHardwareResource"; + public static final String NODE_SOFTWARE_RESOURCE = NS + "NodeSoftwareResource"; + public static final String CONTAINER_RUNTIME = NS + "ContainerRuntime"; + public static final String DOCKER_CONTAINER_RUNTIME = NS + "DockerContainerRuntime"; + public static final String HAS_OPERATING_SYSTEM = NS + "hasOperatingSystem"; + + public static final String HAS_KERNEL_VERSION = NS + "hasKernelVersion"; + public static final String HAS_CONTAINER_RUNTIME = NS + "hasContainerRuntime"; + public static final String NODE_FIELD_DEVICE_ACCESS_RESOURCE = NS + "FieldDeviceAccessResource"; + public static final String FIELD_DEVICE_NAME = NS + "hasFieldDeviceName"; + public static final String FIELD_DEVICE_TYPE = NS + "hasFieldDeviceType"; + + public static final String FIELD_DEVICE_CONNECTION_TYPE = NS + "hasFieldDeviceConnectionType"; + public static final String FIELD_DEVICE_CONNECTION_STRING = NS + "hasFieldDeviceConnectionString"; + public static final String HAS_CPU = NS + "hasCPU"; + + public static final String HAS_MEMORY = NS + "hasMemory"; + + public static final String HAS_DISK = NS + "hasDisk"; + + public static final String HAS_GPU = NS + "hasGpu"; + public static final String NODE_HARDWARE_RESOURCE_CPU = NS + "CpuResource"; + + public static final String HAS_CPU_CORES = NS + "hasCpuCores"; + public static final String HAS_CPU_ARCH = NS + "hasCpuArch"; + public static final String NODE_HARDWARE_RESOURCE_MEM = NS + "MemResource"; + public static final String HAS_MEMORY_TOTAL = NS + "hasMemoryTotal"; + public static final String NODE_HARDWARE_RESOURCE_DISK = NS + "DiskResource"; + public static final String HAS_DISK_TOTAL = NS + "hasDiskTotal"; + public static final String NODE_HARDWARE_RESOURCE_GPU = NS + "GpuResource"; + public static final String HAS_GPU_TYPE = NS + "hasGpuType"; + public static final String HAS_GPU_CORES = NS + "hasGpuCores"; + public static final String HAS_SUPPORTED_ELEMENTS = NS + "hasSupportedElements" ; + public static final String HAS_NODE_BROKER = NS + "hasNodeBroker"; + public static final String NODE_BROKER_DESCRIPTION = NS + "NodeBrokerDescription" ; + public static final String NVIDIA_CONTAINER_RUNTIME = NS + "NvidiaContainerRuntime" ; + public static final String HAS_CUDA_DRIVER_VERSION = NS + "hasCudaDriverVersion"; + + public static final String HAS_CUDA_RUNTIME_VERSION = NS + "hasCudaRuntimeVersion"; + public static final String HAS_FIELD_DEVICE_ACCESS_RESOURCES = NS + "hasFieldDeviceAcessResources"; + public static final String HAS_SOFTWARE_RESOURCES = NS + "hasSoftwareResources"; + + public static final String HAS_HARDWARE_RESOURCES = NS + "hasHardwareResources"; + public static final String HAS_CONTAINER_RUNTIME_SERVER_VERSION = NS + "hasContainerRuntimeServerVersion"; + public static final String HAS_CONTAINER_RUNTIME_API_VERSION = NS + "hasContainerRuntimeApiVersion"; + public static final String HARDWARE_REQUIREMENT = NS + "HardwareRequirement"; + public static final String HAS_GPU_REQUIREMENT = NS + "hasGpuRequirement"; + public static final String HAS_NODE_RESOURCE_PROPERTY = NS + "hasNodeResourceProperty"; + public static final String NODE_RESOURCE_REQUIREMENT = NS + "NodeResourceRequirement"; + public static final String REQUIRES_RESOURCES = NS + "requiresResources"; ++ + public static final String HAS_REQUIRED_FILETYPES = NS + "hasRequiredFiletypes"; + public static final String IS_INTERNALLY_MANAGED = NS + "isInternallyManaged"; + public static final String HAS_CORRESPONDING_ADAPTER_ID = NS + "hasCorrespondingAdapterId"; } diff --cc ui/src/app/core-model/gen/streampipes-model-client.ts index 4e18b90,b168ade..8586b05 --- a/ui/src/app/core-model/gen/streampipes-model-client.ts +++ b/ui/src/app/core-model/gen/streampipes-model-client.ts @@@ -19,7 -19,22 +19,22 @@@ /* tslint:disable */ /* eslint-disable */ // @ts-nocheck - // Generated using typescript-generator version 2.24.612 on 2020-11-25 23:55:36. -// Generated using typescript-generator version 2.27.744 on 2021-01-05 12:29:12. ++// Generated using typescript-generator version 2.27.744 on 2021-02-17 10:58:52. + + export class Element { + elementId: string; + publicElement: boolean; + + static fromData(data: Element, target?: Element): Element { + if (!data) { + return data; + } + const instance = target || new Element(); + instance.elementId = data.elementId; + instance.publicElement = data.publicElement; + return instance; + } + } export class FileMetadata { createdAt: number; @@@ -100,3 -192,3 +192,4 @@@ function __copyObject<T>(object: { [ind function __identity<T>(): (value: T) => T { return value => value; } ++ diff --cc ui/src/app/core-model/gen/streampipes-model.ts index 44fde71,9262cf9..2bdc014 --- a/ui/src/app/core-model/gen/streampipes-model.ts +++ b/ui/src/app/core-model/gen/streampipes-model.ts @@@ -19,10 -19,10 +19,10 @@@ /* tslint:disable */ /* eslint-disable */ // @ts-nocheck - // Generated using typescript-generator version 2.24.612 on 2021-01-09 20:47:18. -// Generated using typescript-generator version 2.27.744 on 2021-02-08 23:06:08. ++// Generated using typescript-generator version 2.27.744 on 2021-02-17 10:58:49. export class AbstractStreamPipesEntity { - "@class": "org.apache.streampipes.model.base.NamedStreamPipesEntity" | "org.apache.streampipes.model.connect.adapter.AdapterDescription" | "org.apache.streampipes.model.connect.adapter.AdapterSetDescription" | "org.apache.streampipes.model.connect.adapter.GenericAdapterSetDescription" | "org.apache.streampipes.model.connect.adapter.SpecificAdapterSetDescription" | "org.apache.streampipes.model.connect.adapter.AdapterStreamDescription" | "org.apache.streampipes.model.connect.adapter. [...] - "@class": "org.apache.streampipes.model.base.AbstractStreamPipesEntity" | "org.apache.streampipes.model.base.NamedStreamPipesEntity" | "org.apache.streampipes.model.connect.adapter.AdapterDescription" | "org.apache.streampipes.model.connect.adapter.AdapterSetDescription" | "org.apache.streampipes.model.connect.adapter.GenericAdapterSetDescription" | "org.apache.streampipes.model.connect.adapter.SpecificAdapterSetDescription" | "org.apache.streampipes.model.connect.adapter.AdapterStr [...] ++ "@class": "org.apache.streampipes.model.base.AbstractStreamPipesEntity" | "org.apache.streampipes.model.base.NamedStreamPipesEntity" | "org.apache.streampipes.model.connect.adapter.AdapterDescription" | "org.apache.streampipes.model.connect.adapter.AdapterSetDescription" | "org.apache.streampipes.model.connect.adapter.GenericAdapterSetDescription" | "org.apache.streampipes.model.connect.adapter.SpecificAdapterSetDescription" | "org.apache.streampipes.model.connect.adapter.AdapterStr [...] elementId: string; static fromData(data: AbstractStreamPipesEntity, target?: AbstractStreamPipesEntity): AbstractStreamPipesEntity { @@@ -695,35 -672,39 +699,68 @@@ export class ColorPickerStaticProperty } } + export class MessagesInfo { + groupId: string; + topicName: string; + + static fromData(data: MessagesInfo, target?: MessagesInfo): MessagesInfo { + if (!data) { + return data; + } + const instance = target || new MessagesInfo(); + instance.topicName = data.topicName; + instance.groupId = data.groupId; + return instance; + } + } + + export class ConsumedMessagesInfo extends MessagesInfo { + consumedMessagesSincePipelineStart: number; + lag: number; + totalMessagesSincePipelineStart: number; + + static fromData(data: ConsumedMessagesInfo, target?: ConsumedMessagesInfo): ConsumedMessagesInfo { + if (!data) { + return data; + } + const instance = target || new ConsumedMessagesInfo(); + super.fromData(data, instance); + instance.totalMessagesSincePipelineStart = data.totalMessagesSincePipelineStart; + instance.consumedMessagesSincePipelineStart = data.consumedMessagesSincePipelineStart; + instance.lag = data.lag; + return instance; + } + } + +export class ContainerRuntime extends UnnamedStreamPipesEntity { + "@class": "org.apache.streampipes.model.node.resources.software.ContainerRuntime" | "org.apache.streampipes.model.node.resources.software.DockerContainerRuntime" | "org.apache.streampipes.model.node.resources.software.NvidiaContainerRuntime"; + apiVersion: string; + serverVersion: string; + + static fromData(data: ContainerRuntime, target?: ContainerRuntime): ContainerRuntime { + if (!data) { + return data; + } + const instance = target || new ContainerRuntime(); + super.fromData(data, instance); + instance.serverVersion = data.serverVersion; + instance.apiVersion = data.apiVersion; + return instance; + } + + static fromDataUnion(data: ContainerRuntimeUnion): ContainerRuntimeUnion { + if (!data) { + return data; + } + switch (data["@class"]) { + case "org.apache.streampipes.model.node.resources.software.DockerContainerRuntime": + return DockerContainerRuntime.fromData(data); + case "org.apache.streampipes.model.node.resources.software.NvidiaContainerRuntime": + return NvidiaContainerRuntime.fromData(data); + } + } +} + export class CorrectionValueTransformationRuleDescription extends ValueTransformationRuleDescription { "@class": "org.apache.streampipes.model.connect.rules.value.CorrectionValueTransformationRuleDescription"; correctionValue: number; @@@ -2865,9 -2602,7 +3000,10 @@@ export class SoftwareResource extends U export class SpDataStream extends NamedStreamPipesEntity { "@class": "org.apache.streampipes.model.SpDataStream" | "org.apache.streampipes.model.SpDataSet"; category: string[]; + correspondingAdapterId: string; + deploymentTargetNodeHostname: string; + deploymentTargetNodeId: string; + deploymentTargetNodePort: number; eventGrounding: EventGrounding; eventSchema: EventSchema; hasEventStreamQualities: EventStreamQualityDefinitionUnion[]; @@@ -2889,10 -2624,8 +3025,11 @@@ instance.measurementCapability = __getCopyArrayFn(MeasurementCapability.fromData)(data.measurementCapability); instance.measurementObject = __getCopyArrayFn(MeasurementObject.fromData)(data.measurementObject); instance.index = data.index; + instance.correspondingAdapterId = data.correspondingAdapterId; instance.category = __getCopyArrayFn(__identity<string>())(data.category); + instance.deploymentTargetNodeId = data.deploymentTargetNodeId; + instance.deploymentTargetNodeHostname = data.deploymentTargetNodeHostname; + instance.deploymentTargetNodePort = data.deploymentTargetNodePort; return instance; } diff --cc ui/src/app/editor/editor.module.ts index 601591f,511678d..f681b5b --- a/ui/src/app/editor/editor.module.ts +++ b/ui/src/app/editor/editor.module.ts @@@ -59,8 -59,7 +59,9 @@@ import {CustomOutputStrategyComponent} import {PropertySelectionComponent} from "./components/output-strategy/property-selection/property-selection.component"; import {UserDefinedOutputStrategyComponent} from "./components/output-strategy/user-defined-output/user-defined-output.component"; import {ConnectModule} from "../connect/connect.module"; +import {MatSliderModule} from "@angular/material/slider"; +import { MigratePipelineProcessorsComponent } from './dialog/migrate-pipeline-processors/migrate-pipeline-processors.component'; + import {PipelineElementTemplateConfigComponent} from "./components/pipeline-element-template-config/pipeline-element-template-config.component"; @NgModule({ imports: [ diff --cc ui/src/app/pipelines/services/pipeline-operations.service.ts index fb5ffea,8cd8ef1..df76ba5 --- a/ui/src/app/pipelines/services/pipeline-operations.service.ts +++ b/ui/src/app/pipelines/services/pipeline-operations.service.ts @@@ -117,16 -120,4 +121,16 @@@ export class PipelineOperationsService showLogs(id) { //this.$state.go("streampipes.pipelinelogs", {pipeline: id}); } + + migratePipelineProcessors(pipelineId: string) { + this.PipelineService.getPipelineById(pipelineId).subscribe(pipeline => { + this.DialogService.open(MigratePipelineProcessorsComponent,{ + panelType: PanelType.SLIDE_IN_PANEL, + title: "Live-Migrate pipeline processors", + data: { + "pipeline": pipeline + } + }); + }) + } - } + } diff --cc ui/src/app/platform-services/platform.module.ts index bc6f440,b5f8882..9d90ac7 --- a/ui/src/app/platform-services/platform.module.ts +++ b/ui/src/app/platform-services/platform.module.ts @@@ -22,8 -22,9 +22,10 @@@ import {PipelineService} from "./apis/p import {PlatformServicesCommons} from "./apis/commons.service"; import {PipelineElementEndpointService} from "./apis/pipeline-element-endpoint.service"; import {FilesService} from "./apis/files.service"; +import {NodeService} from "./apis/node.service"; import {MeasurementUnitsService} from "./apis/measurement-units.service"; + import {PipelineElementTemplateService} from "./apis/pipeline-element-template.service"; + import {PipelineMonitoringService} from "./apis/pipeline-monitoring.service"; @NgModule({ imports: [], @@@ -33,10 -34,11 +35,12 @@@ MeasurementUnitsService, PlatformServicesCommons, PipelineElementEndpointService, + PipelineElementTemplateService, //PipelineTemplateService, PipelineElementService, + PipelineMonitoringService, - PipelineService + PipelineService, - NodeService, ++ NodeService ], entryComponents: [] })
