This is an automated email from the ASF dual-hosted git repository. SvenO3 pushed a commit to branch 4430-chart-settings-not-correctly-migrated-after-event-schema-changes in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 1d8e3c2896ccc138815eec36fffd29af4f233960 Author: Sven Oehler <[email protected]> AuthorDate: Mon May 18 15:56:39 2026 +0200 Add pipeline update migration panel --- .../src/lib/apis/pipeline.service.ts | 19 ++ .../save-pipeline-update-migration.component.html | 191 +++++++++++++++++++++ .../save-pipeline-update-migration.component.scss | 115 +++++++++++++ .../save-pipeline-update-migration.component.ts | 74 ++++++++ .../save-pipeline/save-pipeline.component.html | 22 ++- .../save-pipeline/save-pipeline.component.ts | 60 ++++++- 6 files changed, 477 insertions(+), 4 deletions(-) diff --git a/ui/projects/streampipes/platform-services/src/lib/apis/pipeline.service.ts b/ui/projects/streampipes/platform-services/src/lib/apis/pipeline.service.ts index 1e6ff20a09..6d9a10aaf9 100644 --- a/ui/projects/streampipes/platform-services/src/lib/apis/pipeline.service.ts +++ b/ui/projects/streampipes/platform-services/src/lib/apis/pipeline.service.ts @@ -22,6 +22,7 @@ import { PlatformServicesCommons } from './commons.service'; import { Observable } from 'rxjs'; import { CompactPipeline, + MeasurementUpdateInfo, Message, Pipeline, PipelineElementRecommendationMessage, @@ -100,6 +101,24 @@ export class PipelineService { ); } + performPipelineMigrationPreflight( + pipeline: Pipeline, + ): Observable<MeasurementUpdateInfo[]> { + const pipelineId = pipeline._id; + return this.http + .put( + `${this.apiBasePath}/pipelines/${pipelineId}/pipeline-migration-preflight`, + pipeline, + ) + .pipe( + map(response => { + return (response as any[]).map(p => + MeasurementUpdateInfo.fromData(p), + ); + }), + ); + } + getPipelines(): Observable<Pipeline[]> { return this.http.get(`${this.apiBasePath}/pipelines`).pipe( map(response => { diff --git a/ui/src/app/editor/dialog/save-pipeline/save-pipeline-update-migration/save-pipeline-update-migration.component.html b/ui/src/app/editor/dialog/save-pipeline/save-pipeline-update-migration/save-pipeline-update-migration.component.html new file mode 100644 index 0000000000..dca284192d --- /dev/null +++ b/ui/src/app/editor/dialog/save-pipeline/save-pipeline-update-migration/save-pipeline-update-migration.component.html @@ -0,0 +1,191 @@ +<!-- +~ 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="migration-view" fxLayout="column" fxLayoutAlign="start stretch"> + @if (measurementUpdateInfos.length > 0) { + <div class="migration-summary" fxLayout="column" fxLayoutGap="4px"> + @if (hasCriticalFieldChanges()) { + <h4> + {{ + 'This pipeline cannot be updated because at least one data lake measurement field changed its datatype.' + | translate + }} + </h4> + } @else { + <h4> + {{ + 'This pipeline update requires manual review.' + | translate + }} + </h4> + } + </div> + <div class="migration-infos" fxLayout="column" fxLayoutGap="12px"> + @for (updateInfo of measurementUpdateInfos; track updateInfo) { + <div + class="pipeline-update-info" + data-cy="sp-pipeline-edit-warning" + > + <div class="pipeline-status" fxLayoutAlign="start center"> + @if ( + updateInfo.criticalMeasurementFieldChanges?.length > + 0 + ) { + <mat-icon class="status-icon error-icon"> + error + </mat-icon> + } @else { + <mat-icon class="status-icon warning-icon"> + warning + </mat-icon> + } + <div> + <div class="pipeline-title"> + {{ 'Measurement' | translate }} + <b>{{ updateInfo.measurementName }}</b> + </div> + <div class="pipeline-description"> + @if ( + updateInfo.criticalMeasurementFieldChanges + ?.length > 0 + ) { + {{ + 'This measurement has datatype changes that must be handled before the pipeline can be updated.' + | translate + }} + } @else { + {{ + 'This measurement has charts that need manual review after the pipeline update.' + | translate + }} + } + </div> + </div> + </div> + + @if ( + updateInfo.criticalMeasurementFieldChanges?.length > 0 + ) { + <div class="validation-infos" fxLayout="column"> + <div + class="validation-info validation-info-error" + data-cy="sp-pipeline-measurement-edit-warning" + > + <div + class="validation-info-title" + fxLayoutAlign="start center" + > + <mat-icon class="error-icon"> + error + </mat-icon> + <span> + {{ + 'Critical datatype changes' + | translate + }} + </span> + </div> + <div class="validation-info-message"> + {{ + getCriticalFieldChangesText( + updateInfo.criticalMeasurementFieldChanges + ) + }} + </div> + </div> + </div> + } + + @if (updateInfo.chartSchemaUpdateInfos?.length > 0) { + <div class="affected-charts" fxLayout="column"> + <div + class="affected-charts-title" + fxLayoutAlign="start center" + > + <mat-icon class="affected-charts-icon"> + query_stats + </mat-icon> + {{ 'Affected charts' | translate }} + </div> + @for ( + chartUpdateInfo of updateInfo.chartSchemaUpdateInfos; + track chartUpdateInfo + ) { + <div + class="affected-chart" + data-cy="sp-pipeline-chart-edit-warning" + > + <div + class="affected-chart-title" + fxLayoutAlign="start center" + > + <mat-icon class="warning-icon"> + warning + </mat-icon> + <span> + {{ 'Chart' | translate }} + <b>{{ + chartUpdateInfo.chartTitle + }}</b> + {{ + 'needs manual review after this update.' + | translate + }} + </span> + </div> + @if ( + chartUpdateInfo.affectedFields?.length > + 0 + ) { + <div class="affected-fields"> + <span class="affected-fields-label"> + {{ + 'Deleted fields' | translate + }} + : + </span> + <span> + {{ + getAffectedFieldsText( + chartUpdateInfo.affectedFields + ) + }} + </span> + </div> + } + </div> + } + </div> + } + </div> + } + </div> + } + <div class="migration-actions"> + <button + mat-button + mat-flat-button + color="accent" + [disabled]="hasCriticalFieldChanges()" + (click)="startUpdateEmitter.emit()" + data-cy="btn-update-pipeline-migrate" + > + {{ 'Update pipeline' | translate }} + </button> + </div> +</div> diff --git a/ui/src/app/editor/dialog/save-pipeline/save-pipeline-update-migration/save-pipeline-update-migration.component.scss b/ui/src/app/editor/dialog/save-pipeline/save-pipeline-update-migration/save-pipeline-update-migration.component.scss new file mode 100644 index 0000000000..9786f3e9ca --- /dev/null +++ b/ui/src/app/editor/dialog/save-pipeline/save-pipeline-update-migration/save-pipeline-update-migration.component.scss @@ -0,0 +1,115 @@ +/*! + * 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. + * + */ + +.migration-view { + width: min(100%, 760px); + margin: 0 auto; +} + +.migration-summary { + text-align: center; + margin-bottom: var(--space-md); +} + +.pipeline-description { + color: var(--color-secondary-text); + font-size: var(--font-size-sm); + line-height: var(--line-height-normal); +} + +.migration-infos { + margin-bottom: var(--space-lg); +} + +.pipeline-update-info { + border: 1px solid var(--color-border-subtle); + border-radius: 6px; + padding: var(--space-md); + margin-bottom: var(--space-xs); + background: var(--color-bg-0); +} + +.pipeline-status { + gap: var(--space-sm); +} + +.pipeline-title, +.affected-charts-title { + font-size: var(--font-size-md); + line-height: var(--line-height-normal); +} + +.affected-charts-title { + gap: var(--space-xs); + font-weight: var(--font-weight-bold); +} + +.status-icon { + flex: 0 0 auto; +} + +.warning-icon { + color: var(--color-warning); +} + +.error-icon { + color: var(--color-error); +} + +.validation-infos, +.affected-charts { + margin-top: var(--space-md); + gap: var(--space-sm); +} + +.validation-info, +.affected-chart { + padding: var(--space-md); + border: 1px solid var(--color-border-subtle); + border-left: 3px solid var(--color-warning); + border-radius: 6px; + background: var(--color-bg-0); +} + +.validation-info-error { + border-left-color: var(--color-error); +} + +.validation-info-title, +.affected-chart-title { + gap: var(--space-xs); +} + +.validation-info-message, +.affected-fields { + margin-top: var(--space-xs); + padding-left: calc(24px + var(--space-xs)); + font-size: var(--font-size-sm); + line-height: var(--line-height-normal); +} + +.affected-fields { + display: flex; + align-items: center; + flex-wrap: wrap; + gap: var(--space-xs); +} + +.migration-actions { + text-align: center; +} diff --git a/ui/src/app/editor/dialog/save-pipeline/save-pipeline-update-migration/save-pipeline-update-migration.component.ts b/ui/src/app/editor/dialog/save-pipeline/save-pipeline-update-migration/save-pipeline-update-migration.component.ts new file mode 100644 index 0000000000..94740288af --- /dev/null +++ b/ui/src/app/editor/dialog/save-pipeline/save-pipeline-update-migration/save-pipeline-update-migration.component.ts @@ -0,0 +1,74 @@ +/* + * 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, EventEmitter, Input, Output } from '@angular/core'; +import { + CriticalMeasurementFieldChange, + MeasurementUpdateInfo, +} from '@streampipes/platform-services'; +import { + LayoutAlignDirective, + LayoutDirective, + LayoutGapDirective, +} from '@ngbracket/ngx-layout/flex'; +import { MatButton } from '@angular/material/button'; +import { MatIcon } from '@angular/material/icon'; +import { TranslatePipe } from '@ngx-translate/core'; + +@Component({ + selector: 'sp-save-pipeline-update-migration', + templateUrl: './save-pipeline-update-migration.component.html', + styleUrls: ['./save-pipeline-update-migration.component.scss'], + imports: [ + LayoutDirective, + LayoutAlignDirective, + LayoutGapDirective, + MatButton, + MatIcon, + TranslatePipe, + ], +}) +export class SavePipelineUpdateMigrationComponent { + @Input() + measurementUpdateInfos: MeasurementUpdateInfo[] = []; + + @Output() + startUpdateEmitter: EventEmitter<void> = new EventEmitter<void>(); + + getAffectedFieldsText(affectedFields: string[]): string { + return affectedFields.join(', '); + } + + hasCriticalFieldChanges(): boolean { + return this.measurementUpdateInfos.some( + updateInfo => + updateInfo.criticalMeasurementFieldChanges?.length > 0, + ); + } + + getCriticalFieldChangesText( + criticalFieldChanges: CriticalMeasurementFieldChange[], + ): string { + return criticalFieldChanges + .map( + change => + `${change.runtimeName} (${change.existingType} -> ${change.updatedType})`, + ) + .join(', '); + } +} diff --git a/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.html b/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.html index 1edb1ee34b..51004cc737 100644 --- a/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.html +++ b/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.html @@ -19,7 +19,11 @@ <div class="sp-dialog-container"> <div class="sp-dialog-content padding-20"> <div fxFlex="100" fxLayout="column"> - @if (!operationCompleted && !operationProgress) { + @if ( + !operationCompleted && + !operationProgress && + !pipelineUpdatePreflight + ) { <sp-save-pipeline-settings [currentPipelineName]="pipeline.name" [submitPipelineForm]="submitPipelineForm" @@ -32,6 +36,18 @@ </sp-save-pipeline-settings> } + @if ( + pipelineUpdatePreflight && + !operationCompleted && + !operationProgress + ) { + <sp-save-pipeline-update-migration + [measurementUpdateInfos]="measurementUpdateInfos" + (startUpdateEmitter)="savePipeline(true)" + > + </sp-save-pipeline-update-migration> + } + @if (operationProgress || operationCompleted) { <sp-multi-step-status-indicator [statusIndicators]="statusIndicators" @@ -70,7 +86,9 @@ Open pipeline overview </button> } - @if (!operationCompleted && !operationSuccess) { + @if ( + !operationCompleted && !operationSuccess && !pipelineUpdatePreflight + ) { <button [disabled]=" !submitPipelineForm.valid || diff --git a/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.ts b/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.ts index 6c44869a26..67f16784d6 100644 --- a/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.ts +++ b/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.ts @@ -27,6 +27,7 @@ import { PipelineCanvasMetadata, PipelineCanvasMetadataService, PipelineOperationStatus, + MeasurementUpdateInfo, PipelineService, SpAssetTreeNode, } from '@streampipes/platform-services'; @@ -57,6 +58,7 @@ import { MultiStepStatusIndicatorComponent } from '../../../core-ui/multi-step-s import { MatDivider } from '@angular/material/divider'; import { PipelineStartedStatusComponent } from '../../../core-ui/pipeline/pipeline-started-status/pipeline-started-status.component'; import { MatButton } from '@angular/material/button'; +import { SavePipelineUpdateMigrationComponent } from './save-pipeline-update-migration/save-pipeline-update-migration.component'; @Component({ selector: 'sp-save-pipeline', @@ -72,6 +74,7 @@ import { MatButton } from '@angular/material/button'; LayoutGapDirective, MatButton, TranslatePipe, + SavePipelineUpdateMigrationComponent, ], }) export class SavePipelineComponent implements OnInit { @@ -117,6 +120,8 @@ export class SavePipelineComponent implements OnInit { statusIndicators: StatusIndicator[] = []; finalPipelineOperationStatus: PipelineOperationStatus; pipelineAction: PipelineAction; + pipelineUpdatePreflight = false; + measurementUpdateInfos: MeasurementUpdateInfo[] = []; ngOnInit() { this.storageOptions.updateModeActive = @@ -196,7 +201,13 @@ export class SavePipelineComponent implements OnInit { this.pipelineCanvasMetadata._rev = undefined; } - savePipeline() { + savePipeline(skipPreflight = false) { + if (this.shouldPerformUpdatePreflight(skipPreflight)) { + this.performUpdatePreflight(); + return; + } + + this.pipelineUpdatePreflight = false; let stopPipeline$: Observable<null | PipelineOperationStatus> = of(null); let savePipeline$: Observable<Message> = @@ -218,6 +229,49 @@ export class SavePipelineComponent implements OnInit { this.performStorageOperations(stopPipeline$, savePipeline$); } + shouldPerformUpdatePreflight(skipPreflight: boolean): boolean { + return ( + !skipPreflight && + this.storageOptions.updateModeActive && + this.storageOptions.updateMode !== 'clone' && + this.hasDataLakeSink() + ); + } + + hasDataLakeSink(): boolean { + return this.pipeline.actions.some( + action => + action.appId === + 'org.apache.streampipes.sinks.internal.jvm.datalake', + ); + } + + performUpdatePreflight(): void { + this.operationProgress = true; + this.addStatusIndicator( + this.translateService.instant('Checking pipeline update'), + Status.PROGRESS, + ); + this.pipelineService + .performPipelineMigrationPreflight(this.pipeline) + .subscribe({ + next: updateInfos => { + if (updateInfos.length === 0) { + this.modifyStatusIndicator(Status.SUCCESS); + this.savePipeline(true); + } else { + this.measurementUpdateInfos = updateInfos; + this.pipelineUpdatePreflight = true; + this.operationProgress = false; + this.statusIndicators = []; + } + }, + error: msg => { + this.onFailure(msg); + }, + }); + } + updateId(entity: InvocablePipelineElementUnion) { const lastIdIndex = entity.elementId.lastIndexOf(':'); entity.elementId = @@ -319,7 +373,9 @@ export class SavePipelineComponent implements OnInit { onFailure(msg?: any) { this.operationCompleted = true; this.operationSuccess = false; - this.modifyStatusIndicator(Status.FAILURE); + if (this.statusIndicators.length > 0) { + this.modifyStatusIndicator(Status.FAILURE); + } } showPipelineOperationStatus(
