|
+ [indeterminate]="this.selection.hasValue() && !this.isAllSelected(cluster.jobs)">
|
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/cluster-page/cluster-page.component.scss b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/cluster-page/cluster-page.component.scss
index 9234ae5dc..6c039deca 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/cluster-page/cluster-page.component.scss
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/cluster-page/cluster-page.component.scss
@@ -9,38 +9,8 @@
* either express or implied. Refer to the License for the specific permissions and
* limitations governing your use of the file.
*/
-
-.demo-table {
- width: 100%;
- font-size: 20px;
-}
-
-.mat-cell {
- padding-left: 15px;
- border: 2px solid rgb(141 141 141 / 50%);
- -webkit-background-clip: padding-box;
- background-clip: padding-box;
-}
-
-.mat-header-row {
- background-color: #333;
- color: white;
- font-size: 0.875rem;
- text-transform: uppercase;
- letter-spacing: normal;
-}
-.mat-header-cell {
- padding-left: 15px;
- color: white;
-}
-
-.mat-header-cell:first-of-type, .mat-cell:first-of-type {
- width: 60px;
-}
-
-.mat-row:hover {
- background: linear-gradient(10deg, #f1f2f3, #dce2ff);
-}
+@use '@angular/material' as mat;
+@import "./styles.scss";
.demo-row-is-clicked {
font-weight: bold;
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/cluster-page/cluster-page.component.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/cluster-page/cluster-page.component.ts
index 88dc7b570..92ddbeebe 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/cluster-page/cluster-page.component.ts
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/cluster-page/cluster-page.component.ts
@@ -12,15 +12,15 @@
import {Component} from '@angular/core';
import {ActivatedRoute} from '@angular/router';
-import {BehaviorSubject, combineLatest, forkJoin, Observable, of, throwError} from 'rxjs';
-import {catchError, map, switchMap} from "rxjs/operators";
-import {Job} from "../cluster-list-page/cluster-list-page.model";
-import {ClusterService} from "../../services/cluster.service";
+import {BehaviorSubject, forkJoin, of, throwError} from 'rxjs';
+import {catchError, map, switchMap} from 'rxjs/operators';
+import {Job} from '../cluster-list-page/cluster-list-page.model';
+import {ClusterService} from '../../services/cluster.service';
import {SelectionModel} from '@angular/cdk/collections';
-import {HttpErrorResponse, HttpResponse} from "@angular/common/http";
-import {SnackbarService, SnackBarStatus} from "../../services/snack-bar.service";
-import {MatDialog} from "@angular/material/dialog";
-import {UploadDialogComponent} from "./dialog/upload-dialog.component";
+import {HttpErrorResponse, HttpResponse} from '@angular/common/http';
+import {SnackbarService, SnackBarStatus} from '../../services/snack-bar.service';
+import {MatDialog} from '@angular/material/dialog';
+import {UploadDialogComponent} from './dialog/upload-dialog.component';
export type DialogData = {
targetUrl: string;
@@ -37,11 +37,10 @@ export type DialogData = {
})
export class ClusterPageComponent {
clusterSubject$ = new BehaviorSubject(null);
- clusterId$: Observable = this._route.paramMap.pipe(map((params) => params.get('clusterId')));
- cluster$ = combineLatest([this.clusterId$, this.clusterSubject$]).pipe(switchMap(([clusterId]) => this._clusterService.getCluster(clusterId)));
- vm$ = combineLatest([this.clusterId$, this.cluster$]).pipe(
- map(([clusterId, cl]) => ({clusterId, cl}))
- );
+ cluster$ = this._route.paramMap.pipe(
+ map((params) => params.get('clusterId')),
+ switchMap((clusterId) => this._clusterService.getCluster(clusterId)));
+
displayedColumns: string[] = ['select', 'name', 'type', 'branch', 'pipeline', 'status', 'created'];
selection = new SelectionModel(true, []);
@@ -74,7 +73,7 @@ export class ClusterPageComponent {
this._clusterService.sendJobCommand(clusterId, action,
{
jobIdHex: job.jobIdString,
- pipelineDir: job.jobPipeline,
+ pipelineName: job.jobPipeline,
branch: job.jobPipeline
}).pipe(catchError(err => {
this._snackBarService.showMessage(err.message, SnackBarStatus.FAIL);
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/component/file-upload/file-upload.module.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/component/file-upload/file-upload.module.ts
index ad121b829..b7022de20 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/component/file-upload/file-upload.module.ts
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/component/file-upload/file-upload.module.ts
@@ -1,12 +1,11 @@
import {NgModule} from "@angular/core";
import {MatIconModule} from "@angular/material/icon";
-import {NzMessageService} from "ng-zorro-antd/message";
import {FileUploadComponent} from "./file-upload.component";
import {MatProgressSpinnerModule} from "@angular/material/progress-spinner";
import {CommonModule} from "@angular/common";
import {MatProgressBarModule} from "@angular/material/progress-bar";
import {MatButtonModule} from "@angular/material/button";
-import {DragDropDirective} from "../../../misc/drag-drop.directives";
+import {DragDropDirective} from 'src/app/misc/drag-drop.directives';
import {ProgressComponent} from "./progress/progress.component";
@NgModule({
@@ -23,9 +22,8 @@ import {ProgressComponent} from "./progress/progress.component";
MatButtonModule,
],
providers: [
- NzMessageService
],
- exports: [FileUploadComponent]
+ exports: [FileUploadComponent, DragDropDirective]
})
export class FileUploadModule {
}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-create/pipeline-create.component.html b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-create/pipeline-create.component.html
new file mode 100644
index 000000000..29baa77ad
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-create/pipeline-create.component.html
@@ -0,0 +1,112 @@
+
+
+ Pipeline Configuration
+ This is general pipeline configuration
+
+
+
+
+
+
+ Configuration Type:
+
+
+
+ {{ jobsEnum.PARSER.label }}
+
+ {{ jobsEnum.TRIAGE.label }}
+
+ {{ jobsEnum.PROFILE.label }}
+
+ {{ jobsEnum.INDEX.label }}
+
+
+
+
+
+
+ cloud_upload
+ Drag and drop file here
+ or
+
+
+
+ file_copy
+ {{ file?.name }}
+ [size:{{ formatBytes(file?.size) }}]
+
+ delete
+
+
+
+ pipeline-stepper works!
+
+
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-stepper/pipeline-stepper.component.scss b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-stepper/pipeline-stepper.component.scss
new file mode 100644
index 000000000..e69de29bb
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-stepper/pipeline-stepper.component.spec.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-stepper/pipeline-stepper.component.spec.ts
new file mode 100644
index 000000000..0f72808a0
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-stepper/pipeline-stepper.component.spec.ts
@@ -0,0 +1,59 @@
+import {ComponentFixture, TestBed} from '@angular/core/testing';
+
+import {PipelineStepperComponent} from './pipeline-stepper.component';
+import {FormsModule} from '@angular/forms';
+import {SharedModule} from 'src/app/shared/share.module';
+import {MatCardModule} from '@angular/material/card';
+import {MatDividerModule} from '@angular/material/divider';
+import {MatTooltipModule} from '@angular/material/tooltip';
+import {MatTableModule} from '@angular/material/table';
+import {MatDialogModule} from '@angular/material/dialog';
+import {MatButtonModule} from '@angular/material/button';
+import {MatIconModule} from '@angular/material/icon';
+import {MatFormFieldModule} from '@angular/material/form-field';
+import {MatInputModule} from '@angular/material/input';
+import {AsyncPipe, NgComponentOutlet, NgForOf, NgIf, NgTemplateOutlet} from '@angular/common';
+import {MatSelectModule} from '@angular/material/select';
+import {MatStepperModule} from '@angular/material/stepper';
+import {NoopAnimationsModule} from '@angular/platform-browser/animations';
+
+describe('PipelineStepperComponent', () => {
+ let component: PipelineStepperComponent;
+ let fixture: ComponentFixture;
+
+ beforeEach(async () => {
+ await TestBed.configureTestingModule({ imports: [
+ SharedModule,
+ MatCardModule,
+ MatDividerModule,
+ MatTooltipModule,
+ MatTableModule,
+ MatDialogModule,
+ MatButtonModule,
+ MatIconModule,
+ MatFormFieldModule,
+ MatInputModule,
+ NgIf,
+ NgTemplateOutlet,
+ NgComponentOutlet,
+ AsyncPipe,
+ MatSelectModule,
+ FormsModule,
+ NgForOf,
+ MatStepperModule,
+ NoopAnimationsModule
+ ],
+
+ declarations: [ PipelineStepperComponent ]
+ })
+ .compileComponents();
+
+ fixture = TestBed.createComponent(PipelineStepperComponent);
+ component = fixture.componentInstance;
+ fixture.detectChanges();
+ });
+
+ it('should create', () => {
+ expect(component).toBeTruthy();
+ });
+});
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-stepper/pipeline-stepper.component.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-stepper/pipeline-stepper.component.ts
new file mode 100644
index 000000000..378f88e73
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-stepper/pipeline-stepper.component.ts
@@ -0,0 +1,10 @@
+import { Component} from '@angular/core';
+
+@Component({
+ selector: 'app-pipeline-stepper',
+ templateUrl: './pipeline-stepper.component.html',
+ styleUrls: ['./pipeline-stepper.component.scss']
+})
+export class PipelineStepperComponent {
+
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-submit/pipeline-submit.component.html b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-submit/pipeline-submit.component.html
new file mode 100644
index 000000000..a46de2f4b
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-submit/pipeline-submit.component.html
@@ -0,0 +1,9 @@
+Dispatching pipeline
+
+check_circle1. Create empty pipeline
+
+
+check_circle2. Start all jobs for pipeline
+
+
+
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-submit/pipeline-submit.component.scss b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-submit/pipeline-submit.component.scss
new file mode 100644
index 000000000..e69de29bb
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-submit/pipeline-submit.component.spec.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-submit/pipeline-submit.component.spec.ts
new file mode 100644
index 000000000..4990c1f7b
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-submit/pipeline-submit.component.spec.ts
@@ -0,0 +1,95 @@
+import {ComponentFixture, TestBed, waitForAsync} from '@angular/core/testing';
+
+import {PipelineSubmitComponent} from './pipeline-submit.component';
+import {FormsModule, ReactiveFormsModule} from '@angular/forms';
+import {Router, RouterModule} from '@angular/router';
+import {SharedModule} from 'src/app/shared/share.module';
+import {MatCardModule} from '@angular/material/card';
+import {MatDividerModule} from '@angular/material/divider';
+import {MatTooltipModule} from '@angular/material/tooltip';
+import {MatTableModule} from '@angular/material/table';
+import {MatDialogModule} from '@angular/material/dialog';
+import {MatButtonModule} from '@angular/material/button';
+import {MatIconModule} from '@angular/material/icon';
+import {MatFormFieldModule} from '@angular/material/form-field';
+import {MatInputModule} from '@angular/material/input';
+import {AsyncPipe, NgComponentOutlet, NgForOf, NgIf, NgTemplateOutlet} from '@angular/common';
+import {MatSelectModule} from '@angular/material/select';
+import {MatAutocompleteModule} from '@angular/material/autocomplete';
+import {MatStepperModule} from '@angular/material/stepper';
+import {MatProgressBarModule} from '@angular/material/progress-bar';
+import {MatSlideToggleModule} from '@angular/material/slide-toggle';
+import {FileUploadModule} from 'src/app/cluster/component/file-upload/file-upload.module';
+import {NoopAnimationsModule} from '@angular/platform-browser/animations';
+import {ClusterPipelineService} from 'src/app/services/cluster.pipeline.service';
+import {of} from 'rxjs';
+import {RouterTestingModule} from '@angular/router/testing';
+
+describe('PipelineSubmitComponent', () => {
+ let component: PipelineSubmitComponent;
+ let fixture: ComponentFixture;
+ let clusterPipelineService: jasmine.SpyObj;
+ let router: Router;
+
+ beforeEach(waitForAsync(() => {
+ TestBed.configureTestingModule({
+ imports: [
+ ReactiveFormsModule,
+ RouterModule,
+ SharedModule,
+ MatCardModule,
+ MatDividerModule,
+ MatTooltipModule,
+ MatTableModule,
+ MatDialogModule,
+ MatButtonModule,
+ MatIconModule,
+ MatFormFieldModule,
+ MatInputModule,
+ NgIf,
+ NgTemplateOutlet,
+ NgComponentOutlet,
+ AsyncPipe,
+ MatSelectModule,
+ FormsModule,
+ NgForOf,
+ MatAutocompleteModule,
+ MatStepperModule,
+ MatProgressBarModule,
+ MatSlideToggleModule,
+ FileUploadModule,
+ RouterTestingModule.withRoutes([]),
+ NoopAnimationsModule
+ ],
+ declarations: [PipelineSubmitComponent],
+ providers: [
+ {
+ provide: ClusterPipelineService,
+ useValue: jasmine.createSpyObj('ClusterPipelineService', ['createEmptyPipeline', 'startAllPipelines', 'getAllPipelines', 'getPipelines'])
+ }
+ ]
+ })
+ .compileComponents();
+ router = TestBed.inject(Router);
+ spyOn(router, 'getCurrentNavigation').and.returnValue({
+ extras: {state: {data: {clusterId: 'test1', pipelineName: 'foo-pipe', branch: 'testbranch'}}},
+ id: 0,
+ initialUrl: null,
+ extractedUrl: null,
+ trigger: null,
+ previousNavigation: null
+ });
+
+
+ fixture = TestBed.createComponent(PipelineSubmitComponent);
+ component = fixture.componentInstance;
+ clusterPipelineService = TestBed.inject(ClusterPipelineService) as jasmine.SpyObj;
+ clusterPipelineService.createEmptyPipeline = jasmine.createSpy('createEmptyPipeline').and.returnValue(of({}));
+ fixture.detectChanges();
+ }));
+
+ it('should create', () => {
+
+ expect(component).toBeTruthy();
+ });
+});
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-submit/pipeline-submit.component.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-submit/pipeline-submit.component.ts
new file mode 100644
index 000000000..8771a5757
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline-submit/pipeline-submit.component.ts
@@ -0,0 +1,49 @@
+import {Component, inject, OnInit} from '@angular/core';
+import {Router} from '@angular/router';
+import {ClusterPipelineService} from 'src/app/services/cluster.pipeline.service';
+
+export type PipelineSubmitState = {
+ clusterId: string;
+ pipelineName: string;
+ branch: string;
+ showStartAllJob?: boolean;
+ profileName?: string;
+ file?: any;
+ jobs?: string[]
+}
+
+@Component({
+ selector: 'app-pipeline-submit',
+ templateUrl: './pipeline-submit.component.html',
+ styleUrls: ['./pipeline-submit.component.scss']
+})
+export class PipelineSubmitComponent implements OnInit {
+ private _route = inject(Router);
+ private _clusterPipelineService = inject(ClusterPipelineService);
+ loadingEmpty = true;
+ loadingStartAllJob = false;
+ state: PipelineSubmitState;
+ finishedSubmitAllJob = false;
+
+ constructor() {
+ const navigation = this._route.getCurrentNavigation();
+ this.state = navigation?.extras.state.data as PipelineSubmitState;
+ }
+
+ ngOnInit(): void {
+ this._clusterPipelineService.createEmptyPipeline(this.state.clusterId, this.state.pipelineName, this.state.branch).subscribe(() => {
+ this.loadingEmpty = false;
+ this.loadingStartAllJob = true;
+ if (this.state.showStartAllJob) {
+ this.loadingStartAllJob = true;
+ this._clusterPipelineService.startAllPipelines(this.state.clusterId, this.state.pipelineName, this.state.jobs, this.state.branch, this.state.profileName, this.state.file).subscribe(
+ () => {
+ this.loadingStartAllJob = false;
+ this.finishedSubmitAllJob = true;
+ }
+ )
+ }
+ })
+ }
+
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline.model.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline.model.ts
new file mode 100644
index 000000000..9e8261a5a
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipeline.model.ts
@@ -0,0 +1,20 @@
+/*
+ * Copyright 2020 - 2022 Cloudera. All Rights Reserved.
+ *
+ * This file is licensed 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.
+ *
+ * This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. Refer to the License for the specific permissions and
+ * limitations governing your use of the file.
+ */
+
+
+export type PipelineModel = {
+ id: string;
+ name: string;
+ clusterName: string;
+ date: string;
+ user: string;
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipelines.component.html b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipelines.component.html
new file mode 100644
index 000000000..8b364a7fc
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipelines.component.html
@@ -0,0 +1,89 @@
+
+
+
+
+ Pipeline Chains
+
+
+
+
+
+
+
+
+ Pipeline list
+
+
+ | Name |
+
+ valve
+ {{ tablePipeline.name }}
+ |
+
+
+
+
+ Cluster |
+ {{ tablePipeline.clusterName }} |
+
+
+
+
+ Date |
+ {{ tablePipeline.date }} |
+
+
+
+
+ User |
+ {{ tablePipeline.user }}
+ |
+
+
+
+
+ Action |
+
+
+ |
+
+
+ |
+ |
+
+
+
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipelines.component.scss b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipelines.component.scss
new file mode 100644
index 000000000..61292046d
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipelines.component.scss
@@ -0,0 +1,89 @@
+/*!
+ * Copyright 2020 - 2022 Cloudera. All Rights Reserved.
+ *
+ * This file is licensed 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.
+ *
+ * This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. Refer to the License for the specific permissions and
+ * limitations governing your use of the file.
+ */
+@use '@angular/material' as mat;
+@import "./styles.scss";
+
+.panel-class {
+ margin-top: 30px !important;
+}
+
+.my-form-field .mat-form-field-wrapper {
+ flex: 0 0 60%;
+}
+
+::ng-deep .mat-card .mat-card-title-group {
+ div:first-child {
+ flex: 1 0 50%;
+ }
+}
+
+.mat-card {
+ .mat-card-title-group {
+ margin-bottom: 10px;
+ margin-left: 15px;
+ margin-right: 15px;
+ }
+
+ .mat-card-header {
+ display: block;
+ }
+}
+
+.chain-delete-btn {
+ color: mat.get-color-from-palette(mat.define-palette(mat.$red-palette), 300);
+}
+.chain-delete-btn:hover {
+ color: mat.get-color-from-palette(mat.define-palette(mat.$red-palette), 600);
+}
+
+.chain-open-btn {
+ color: mat.get-color-from-palette($theme-primary, 300);
+ .chain-open-btn-square {
+ background-color: mat.get-color-from-palette($theme-primary, 300);
+ width: 22px;
+ height: 22px;
+ border-radius: 6px;
+ font-size: 22px;
+ margin-right: 5px;
+ .mat-icon {
+ color: white;
+ }
+ }
+}
+
+.chain-open-btn:hover {
+ color: mat.get-color-from-palette($theme-primary, 600);
+
+ .chain-open-btn-square {
+ background-color: mat.get-color-from-palette($theme-primary, 600);
+ }
+}
+.title-action-wrapper {
+ display: flex;
+ flex: 1 0 50%;
+ justify-content: flex-end
+}
+.create-new-pipeline {
+ min-height: 42px;
+ min-width: 124px;
+ font-size: 18px;
+ border-radius: 12px;
+ text-transform: uppercase;
+ letter-spacing: normal;
+}
+
+.mat-divider-wrapper {
+ margin: 0 5px;
+ .mat-divider {
+ height: 50%;
+ }
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipelines.component.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipelines.component.ts
new file mode 100644
index 000000000..bc205eb24
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipelines.component.ts
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2020 - 2022 Cloudera. All Rights Reserved.
+ *
+ * This file is licensed 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.
+ *
+ * This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. Refer to the License for the specific permissions and
+ * limitations governing your use of the file.
+ */
+
+import {Component, inject} from '@angular/core';
+import {ClusterPipelineService} from 'src/app/services/cluster.pipeline.service';
+
+@Component({
+ selector: 'app-chain-list-page',
+ templateUrl: './pipelines.component.html',
+ styleUrls: ['./pipelines.component.scss'],
+})
+export class PipelinesComponent {
+ //Injects and services
+ private _pipelineService = inject(ClusterPipelineService);
+
+ readonly columns = ['name', 'cluster', 'date', 'user', 'action'];
+
+ pipelines$ = this._pipelineService.getAllPipelines();
+
+
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipelines.module.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipelines.module.ts
new file mode 100644
index 000000000..619356371
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/cluster/pipelines/pipelines.module.ts
@@ -0,0 +1,73 @@
+/*
+ * Copyright 2020 - 2022 Cloudera. All Rights Reserved.
+ *
+ * This file is licensed 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.
+ *
+ * This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. Refer to the License for the specific permissions and
+ * limitations governing your use of the file.
+ */
+
+import {NgModule} from '@angular/core';
+import {FormsModule, ReactiveFormsModule} from '@angular/forms';
+import {RouterModule} from '@angular/router';
+import {SharedModule} from 'src/app/shared/share.module';
+import {MatCardModule} from '@angular/material/card';
+import {MatDividerModule} from '@angular/material/divider';
+import {MatTooltipModule} from '@angular/material/tooltip';
+import {MatTableModule} from '@angular/material/table';
+import {MatDialogModule} from '@angular/material/dialog';
+import {MatButtonModule} from '@angular/material/button';
+import {MatIconModule} from '@angular/material/icon';
+import {MatFormFieldModule} from '@angular/material/form-field';
+import {MatInputModule} from '@angular/material/input';
+import {AsyncPipe, NgComponentOutlet, NgForOf, NgIf, NgTemplateOutlet} from '@angular/common';
+import {MatSelectModule} from '@angular/material/select';
+import {MatAutocompleteModule} from '@angular/material/autocomplete';
+import {PipelinesComponent} from 'src/app/cluster/pipelines/pipelines.component';
+import {PipelineStepperComponent} from 'src/app/cluster/pipelines/pipeline-stepper/pipeline-stepper.component';
+import {MatStepperModule} from '@angular/material/stepper';
+import {PipelineCreateComponent} from './pipeline-create/pipeline-create.component';
+import {PipelineSubmitComponent} from './pipeline-submit/pipeline-submit.component';
+import {MatProgressBarModule} from '@angular/material/progress-bar';
+import {MatSlideToggleModule} from '@angular/material/slide-toggle';
+import {FileUploadModule} from 'src/app/cluster/component/file-upload/file-upload.module';
+import {MatProgressSpinnerModule} from '@angular/material/progress-spinner';
+
+
+@NgModule({
+ declarations: [PipelinesComponent, PipelineStepperComponent, PipelineCreateComponent, PipelineSubmitComponent],
+ imports: [
+ ReactiveFormsModule,
+ RouterModule,
+ SharedModule,
+ MatCardModule,
+ MatDividerModule,
+ MatTooltipModule,
+ MatTableModule,
+ MatDialogModule,
+ MatButtonModule,
+ MatIconModule,
+ MatFormFieldModule,
+ MatInputModule,
+ NgIf,
+ NgTemplateOutlet,
+ NgComponentOutlet,
+ AsyncPipe,
+ MatSelectModule,
+ FormsModule,
+ NgForOf,
+ MatAutocompleteModule,
+ MatStepperModule,
+ MatProgressBarModule,
+ MatSlideToggleModule,
+ FileUploadModule,
+ MatProgressSpinnerModule
+ ],
+ providers: [],
+ exports: [PipelinesComponent]
+})
+export class PipelinesModule {
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/components/popup/json-editor-popup/json-editor-popup.component.spec.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/components/popup/json-editor-popup/json-editor-popup.component.spec.ts
index b0259ddac..bed65ec0f 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/components/popup/json-editor-popup/json-editor-popup.component.spec.ts
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/components/popup/json-editor-popup/json-editor-popup.component.spec.ts
@@ -3,7 +3,7 @@ import {ComponentFixture, TestBed, waitForAsync} from '@angular/core/testing';
import {JsonEditorPopupComponent} from './json-editor-popup.component';
import {MAT_DIALOG_DATA, MatDialogModule, MatDialogRef} from "@angular/material/dialog";
import {MockComponent} from "ng-mocks";
-import {AdvancedEditorComponent} from "../../../chain-page/components/parser/advanced-editor/advanced-editor.component";
+import {AdvancedEditorComponent} from 'src/app/chain-page/components/parser/advanced-editor/advanced-editor.component';
describe('JsonEditorPopupComponent', () => {
let component: JsonEditorPopupComponent;
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/chain-list-page.service.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/chain-list-page.service.ts
index 63f5342ce..76171a089 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/chain-list-page.service.ts
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/chain-list-page.service.ts
@@ -14,7 +14,7 @@ import {HttpClient, HttpParams} from '@angular/common/http';
import {Injectable} from '@angular/core';
import {ChainModel, ChainOperationalModel} from '../chain-list-page/chain.model';
-import {getHttpParams} from "../shared/service.utils";
+import {getHttpParams} from '../shared/service.utils';
@Injectable({
providedIn: 'root'
@@ -30,26 +30,21 @@ export class ChainListPageService {
public createChain(chain: ChainOperationalModel, pipeline: string = null) {
const httpParams: HttpParams = getHttpParams(pipeline);
-
return this._http.post(ChainListPageService.BASE_URL + 'chains', chain, {params: httpParams});
}
public getChains(pipeline: string = null) {
const httpParams: HttpParams = getHttpParams(pipeline);
-
return this._http.get(ChainListPageService.BASE_URL + 'chains', {params: httpParams});
}
public deleteChain(chainId: string, pipeline: string = null) {
const httpParams: HttpParams = getHttpParams(pipeline);
-
return this._http.delete(ChainListPageService.BASE_URL + 'chains/' + chainId, {params: httpParams});
}
public getPipelines() {
const httpParams: HttpParams = getHttpParams(null);
-
return this._http.delete(ChainListPageService.BASE_URL + 'pipeline', {params: httpParams});
}
-
}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/cluster.pipeline.service.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/cluster.pipeline.service.ts
new file mode 100644
index 000000000..9f175e258
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/cluster.pipeline.service.ts
@@ -0,0 +1,47 @@
+import {inject, Injectable} from '@angular/core';
+import {HttpClient, HttpRequest} from '@angular/common/http';
+import {PipelineModel} from 'src/app/cluster/pipelines/pipeline.model';
+import {RequestBody} from 'src/app/cluster/cluster-list-page/cluster-list-page.model';
+
+@Injectable({
+ providedIn: 'root'
+})
+export class ClusterPipelineService {
+
+ static readonly BASE_URL = '/api/v1/clusters';
+ private _http = inject(HttpClient);
+
+
+ public createEmptyPipeline(clusterId: string, pipelineName: string, branch: string = 'main') {
+ const body: RequestBody = {
+ pipelineName,
+ branch
+ }
+ return this._http.post(ClusterPipelineService.BASE_URL + `/${clusterId}/pipelines`,body);
+ }
+
+ public startAllPipelines(clusterId: string, pipelineName: string, jobs: string[], branch: string, profileName: string, file: any) {
+ const fd = new FormData();
+ const blobJson = new Blob([JSON.stringify({
+ jobs,
+ profileName,
+ branch,
+ })], {
+ type: 'application/json'
+ });
+ const blobFile = new Blob([file], {
+ type: "multipart/form-data"
+ })
+ fd.append('payload', blobFile);
+ fd.append('body', blobJson);
+ const req = new HttpRequest('POST', ClusterPipelineService.BASE_URL + `/${clusterId}/pipelines/${pipelineName}/start`, fd);
+ return this._http.request(req);
+ }
+
+
+ public getAllPipelines = () =>
+ this._http.get(ClusterPipelineService.BASE_URL + '/pipelines');
+
+ public getPipelines = (clusterId : string | number) =>
+ this._http.get(`${ClusterPipelineService.BASE_URL}/${clusterId}/pipelines`);
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/cluster.service.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/cluster.service.ts
index 18830d5d9..d71a00052 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/cluster.service.ts
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/cluster.service.ts
@@ -1,6 +1,6 @@
-import {Injectable} from "@angular/core";
-import {HttpClient, HttpResponse} from "@angular/common/http";
-import {ClusterModel, RequestBody} from "../cluster/cluster-list-page/cluster-list-page.model";
+import {Injectable} from '@angular/core';
+import {HttpClient, HttpResponse} from '@angular/common/http';
+import {ClusterModel, RequestBody} from '../cluster/cluster-list-page/cluster-list-page.model';
@Injectable({
providedIn: 'root'
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/pipeline.service.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/pipeline.service.ts
index d598f25ae..3e6ec0cb4 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/pipeline.service.ts
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/services/pipeline.service.ts
@@ -13,10 +13,8 @@
import {HttpClient, HttpParams} from '@angular/common/http';
import {Injectable} from '@angular/core';
-import {map, take} from "rxjs/operators";
-import {select, Store} from "@ngrx/store";
-import {ChainListPageState, getSelectedPipeline} from "../chain-list-page/chain-list-page.reducers";
-import {getHttpParams} from "../shared/service.utils";
+import {map} from 'rxjs/operators';
+import {getHttpParams} from '../shared/service.utils';
@Injectable({
providedIn: 'root'
@@ -25,17 +23,13 @@ export class PipelineService {
static readonly BASE_URL = '/api/v1/pipeline';
- currentPipeline$ = this._store.pipe(select(getSelectedPipeline));
-
constructor(
private _http: HttpClient,
- private _store: Store
) {
}
public getPipelines() {
const httpParams: HttpParams = getHttpParams(null);
-
return this._http.get(PipelineService.BASE_URL, {params: httpParams})
.pipe(map(strArr => strArr.map(pipelineName => {
return pipelineName
@@ -71,12 +65,4 @@ export class PipelineService {
return pName
})));
}
-
- public getCurrentPipeline() {
- let currentPipeline: string;
- this.currentPipeline$.pipe(take(1))
- .subscribe(pipelineName => currentPipeline = pipelineName);
- return currentPipeline;
- }
-
}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/confirm-delete-dialog/confirm-delete-dialog.component.html b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/confirm-delete-dialog/confirm-delete-dialog.component.html
new file mode 100644
index 000000000..0847654cf
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/confirm-delete-dialog/confirm-delete-dialog.component.html
@@ -0,0 +1,11 @@
+Confirm delete
+
+ Are you sure you want to delete this item?
+
+
+
+
+
+
+
+
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/confirm-delete-dialog/confirm-delete-dialog.component.scss b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/confirm-delete-dialog/confirm-delete-dialog.component.scss
new file mode 100644
index 000000000..e69de29bb
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/confirm-delete-dialog/confirm-delete-dialog.component.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/confirm-delete-dialog/confirm-delete-dialog.component.ts
new file mode 100644
index 000000000..99506827b
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/confirm-delete-dialog/confirm-delete-dialog.component.ts
@@ -0,0 +1,35 @@
+import {Component, inject} from '@angular/core';
+import {Observable} from 'rxjs';
+import {MAT_DIALOG_DATA, MatDialogRef} from '@angular/material/dialog';
+import {SnackbarService, SnackBarStatus} from 'src/app/services/snack-bar.service';
+
+export type DeleteDialogData = {
+ action: () => Observable;
+}
+
+@Component({
+ selector: 'app-confirm-delete-dialog',
+ templateUrl: './confirm-delete-dialog.component.html',
+ styleUrls: ['./confirm-delete-dialog.component.scss']
+})
+export class ConfirmDeleteDialogComponent {
+ private _data = inject(MAT_DIALOG_DATA) as DeleteDialogData;
+ private _dialogRef = inject(MatDialogRef);
+ private _snackBar = inject(SnackbarService);
+ loading = false;
+
+ delete() {
+ this.loading = true;
+ this._data.action().subscribe(
+ result => {
+ this.loading = false;
+ this._snackBar.showMessage('Successfully deleted', SnackBarStatus.SUCCESS);
+ this._dialogRef.close(result);
+ },
+ (error) => {
+ this.loading = false;
+ this._snackBar.showMessage('Failed to delete ' + error?.message);
+ }
+ );
+ }
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.html b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.html
index 6475284bc..417db53f8 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.html
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.html
@@ -1,5 +1,5 @@
-
-
+
+
{{button.label}}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.scss b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.scss
index 8b1378917..b773b7bbc 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.scss
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.scss
@@ -1 +1,14 @@
+@use '@angular/material' as mat;
+@import './styles.scss';
+
+.mat-button-toggle-disabled {
+ &.mat-button-toggle-appearance-standard {
+ background-color: mat.get-color-from-palette($theme-primary,300);
+ color: mat.get-color-from-palette($theme-primary,100);
+ }
+ &.mat-button-toggle-appearance-standard:hover {
+ background-color: mat.get-color-from-palette($theme-primary,300);
+ color: mat.get-color-from-palette($theme-primary,100);
+ }
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.spec.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.spec.ts
index 48101c03b..972a16d5a 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.spec.ts
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.spec.ts
@@ -28,12 +28,6 @@ describe('MultiButtonComponent', () => {
expect(component).toBeTruthy();
});
- it('should have buttons as per @Input', () => {
- component.buttons = [ {label: "Test1", value: "Value1"}, {label: "Test2", value: "Value2"} ];
- expect(component.buttons.length).toBe(2);
- expect(component.buttons.map(button => button.label)).toEqual(['Test1', 'Test2']);
- });
-
it('should have buttons as default', () => {
expect(component.buttons.length).toBe(3);
expect(component.buttons.map(button => button.label)).toEqual(['Archive', 'Git', 'Manual']);
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.ts
index 217d2d0c4..8d6777762 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.ts
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.component.ts
@@ -1,9 +1,10 @@
import {Component, EventEmitter, Input, Output, ViewChild} from '@angular/core';
import {MatButtonToggleGroup} from "@angular/material/button-toggle";
-type MultiButton = {
+export type MultiButton = {
label: string,
value: string,
+ disabled: boolean
}
@Component({
@@ -15,10 +16,11 @@ type MultiButton = {
})
export class MultiButtonComponent {
@Input() buttons: MultiButton[] = [
- {label: 'Archive', value: 'Archive'},
- {label: 'Git', value: 'Git'},
- {label: 'Manual', value: 'Manual'},
+ {label: 'Archive', value: 'Archive', disabled: false},
+ {label: 'Git', value: 'Git', disabled: false},
+ {label: 'Manual', value: 'Manual', disabled: false},
];
+ @Input() defaultValue: string;
@Output() valueChange = new EventEmitter();
@@ -27,7 +29,6 @@ export class MultiButtonComponent {
onValueChange() {
this.valueChange.emit(this.group.value);
}
-
}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.module.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.module.ts
deleted file mode 100644
index 8ac2f8ef2..000000000
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/components/multibutton/multi-button.module.ts
+++ /dev/null
@@ -1,18 +0,0 @@
-import {NgModule} from "@angular/core";
-import {MultiButtonComponent} from "./multi-button.component";
-import {MatButtonToggleModule} from "@angular/material/button-toggle";
-import {CommonModule} from "@angular/common";
-
-@NgModule({
- declarations: [
- MultiButtonComponent
- ],
- imports: [
- MatButtonToggleModule,
- CommonModule,
- ],
- providers: [],
- exports: [MultiButtonComponent]
-})
-export class MultiButtonModule {
-}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/pipes/sort.pipe.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/pipes/sort.pipe.ts
new file mode 100644
index 000000000..3df72793b
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/pipes/sort.pipe.ts
@@ -0,0 +1,20 @@
+import {Pipe, PipeTransform} from "@angular/core";
+import {sortBy, SortOrder} from '../utils';
+
+@Pipe({name: "sortPipe"})
+export class SortPipe implements PipeTransform {
+ transform(value: T[],column : keyof T, order:SortOrder = 'asc'): T[] {
+ if (!value || value.length <= 1) {
+ return value;
+ }
+ if (!column || column === '') {
+ if (order === 'asc') {
+ return value.sort();
+ } else {
+ return value.sort().reverse();
+ }
+ }
+ return sortBy(value, column, order);
+ }
+
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/service.utils.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/service.utils.ts
index 69e0f6129..ee1fa89cc 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/service.utils.ts
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/service.utils.ts
@@ -12,12 +12,20 @@
import {HttpParams} from "@angular/common/http";
-export function getHttpParams(pipeline: string) {
- const httpParams: HttpParams = new HttpParams();
+export function getHttpParams(map: string | Map | {[key:string] : string}): HttpParams;
- if (pipeline) {
- return httpParams.set('pipelineName', pipeline);
+export function getHttpParams(arg: any) {
+ if (arg === null || arg === undefined) {
+ return new HttpParams();
}
-
- return httpParams
+ if (typeof arg === 'string' && arg) {
+ return new HttpParams().set('pipelineName', arg);
+ }
+ if (arg instanceof Map) {
+ return arg;
+ }
+ if (typeof arg === 'object') {
+ return new HttpParams({fromObject: arg});
+ }
+ return new HttpParams();
}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/share.module.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/share.module.ts
new file mode 100644
index 000000000..35676d907
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/share.module.ts
@@ -0,0 +1,18 @@
+import {NgModule} from '@angular/core';
+import {CommonModule} from '@angular/common';
+import {SortPipe} from './pipes/sort.pipe';
+import {MultiButtonComponent} from './components/multibutton/multi-button.component';
+import {MatButtonToggleModule} from '@angular/material/button-toggle';
+import { ConfirmDeleteDialogComponent } from './components/confirm-delete-dialog/confirm-delete-dialog.component';
+import {A11yModule} from '@angular/cdk/a11y';
+import {MatDialogModule} from '@angular/material/dialog';
+import {MatButtonModule} from '@angular/material/button';
+import {MatProgressSpinnerModule} from '@angular/material/progress-spinner';
+import {MatProgressBarModule} from '@angular/material/progress-bar';
+
+@NgModule({
+ imports: [CommonModule, MatButtonToggleModule, A11yModule, MatDialogModule, MatButtonModule, MatProgressSpinnerModule, MatProgressBarModule],
+ declarations: [ SortPipe, MultiButtonComponent, ConfirmDeleteDialogComponent ],
+ exports: [ SortPipe, MultiButtonComponent, ConfirmDeleteDialogComponent ]
+})
+export class SharedModule { }
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/utils.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/utils.ts
index 7510f7b2c..e2581fac6 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/utils.ts
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/utils.ts
@@ -1,3 +1,7 @@
+import {Observable, of} from 'rxjs';
+import {catchError, switchMap, take} from 'rxjs/operators';
+import {AsyncValidatorFn, FormArray, FormControl, FormGroup, ValidationErrors, ValidatorFn} from '@angular/forms';
+
/**
* Convert a string or ArrayBuffer to a string.
* If the input is null, return an empty string.
@@ -92,3 +96,70 @@ export function findValues(obj: object, searchKey: string): T[] {
return values;
}, []);
}
+
+export function sortBy(arr: T[], field: keyof T, order: SortOrder = 'asc'): T[] {
+ return arr.slice().sort((a: T, b: T) => {
+ const valueA = a[field];
+ const valueB = b[field];
+
+ if (typeof valueA === 'string' || typeof valueA === 'number') {
+ const comparisonValue = valueA > valueB ? 1 : -1;
+ const sortMultiplier = order === 'asc' ? 1 : -1;
+ return comparisonValue * sortMultiplier;
+ }
+ throw new Error(`Unsupported data type for field '${String(field)}'.`);
+ });
+}
+
+export const uniqueAsyncValidator: UniqueAsyncValidatorType = (existValues, column?) => (control) => existValues.pipe(
+ switchMap(values => unique(control.value, column)(values) ? of({uniqueValue: true}) : of(null)),
+ catchError(() => of({httpError: true})),
+ take(1)
+);
+
+export const uniqueValidator: UniqueValidatorType = (existValues, column?) => (control): ValidationErrors | null => {
+ for (const existValue of existValues) {
+ if (!!column && existValue[column] === control.value) {
+ return {uniqueValue: true};
+ } else if (existValue === control.value) {
+ return {uniqueValue: true};
+ }
+ }
+ return null
+}
+
+export function isExist(items: T[], value: string, column: keyof T) {
+ return items.some(item => {
+ const itemValue = column ? String(item[column]).toLowerCase() : String(item).toLowerCase();
+ return itemValue === value.toLowerCase();
+ });
+}
+
+
+export const changeStateFn = (value: any, key?: keyof T, newValue?: any) =>
+ (state: T[]): T[] => state.map(mapState(key, value, newValue)).filter(item => item !== null);
+
+function mapState(key: keyof T, value: any, newValue?: any) {
+ return (item: T) => {
+ const condition = key ? item[key] === value : item === value;
+ if (condition) {
+ return newValue ? newValue : null;
+ }
+ return item;
+ };
+}
+
+export const unique: UniqueFunction = (value, column) => (items) => isExist(items, value, column);
+
+export type UniqueAsyncValidatorType = (existingValues: Observable, column?: keyof T) => AsyncValidatorFn;
+export type UniqueValidatorType = (existingValues: T[], column?: keyof T) => ValidatorFn;
+export type UniqueFunction = (value: string, column?: keyof T) => (items: T[]) => boolean;
+export type SortOrder = 'asc' | 'desc';
+
+export type TypedFormControls> = {
+ [K in keyof T]-?: T[K] extends (infer R)[]
+ ? FormArray ? FormGroup> : FormControl>
+ : T[K] extends Record
+ ? FormGroup>
+ : FormControl;
+};
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/validators/chain.async.validator.ts b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/validators/chain.async.validator.ts
new file mode 100644
index 000000000..f169287e1
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/app/shared/validators/chain.async.validator.ts
@@ -0,0 +1,24 @@
+import {AbstractControl, AsyncValidator, ValidationErrors} from '@angular/forms';
+import {Observable, of} from 'rxjs';
+import {catchError, switchMap} from 'rxjs/operators';
+import {unique} from 'src/app/shared/utils';
+import {ChainListPageService} from 'src/app/services/chain-list-page.service';
+import {Injectable} from '@angular/core';
+
+@Injectable({
+ providedIn: 'root'
+})
+export class ChainAsyncValidator implements AsyncValidator {
+ constructor(private _service: ChainListPageService
+ ) {
+ }
+
+ validate(control: AbstractControl):
+ Promise | Observable {
+ const uniqueFunc = unique(control.value, 'name');
+ return this._service.getChains().pipe(
+ switchMap(chains => uniqueFunc(chains) ? of({uniqueValue: true}) : of(null)),
+ catchError(() => of({httpError: true}))
+ );
+ }
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/styles.scss b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/styles.scss
index 9674bd221..1142d2e09 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/styles.scss
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/styles.scss
@@ -32,7 +32,8 @@ $parser-chaining-theme: mat.define-light-theme((
primary: $parser-chaining-primary,
accent: $parser-chaining-accent,
warn: $parser-chaining-warn,
- )
+ ),
+ typography: $typography
));
// Include theme styles for core and each component used in your app.
@@ -53,6 +54,7 @@ $theme-warn: mat.define-palette(mat.$yellow-palette);
.ant-card-body {
padding: 16px;
}
+
.ant-card-head {
background: var(--card-head-bg);
}
@@ -71,9 +73,11 @@ $theme-warn: mat.define-palette(mat.$yellow-palette);
.ant-table-tbody > tr:hover:not(.ant-table-expanded-row) > td {
background: var(--background);
}
+
.ant-table-thead > tr > th {
border-bottom: 1px solid var(--light-border);
}
+
.ant-table-tbody > tr > td {
border-bottom: 1px solid var(--light-separator);
}
@@ -112,6 +116,7 @@ $theme-warn: mat.define-palette(mat.$yellow-palette);
.ant-pagination-disabled:focus .ant-pagination-item-link {
border-color: var(--light-border);
}
+
// Menu
.ant-menu-item > a:hover {
@@ -150,7 +155,7 @@ $theme-warn: mat.define-palette(mat.$yellow-palette);
}
.ant-select-selection {
- border:1px solid var(--input-border-color);
+ border: 1px solid var(--input-border-color);
}
.parser-chain .ant-card-head-title {
@@ -160,40 +165,41 @@ $theme-warn: mat.define-palette(mat.$yellow-palette);
}
.success-snackbar {
- background: rgb(0, 110, 37);
- color: white;
+ background: mat.get-color-from-palette($theming-material-components-success, 600);
+ color: mat.get-contrast-color-from-palette($theming-material-components-success, 600);
font-weight: bold;
margin-bottom: 5px;
- letter-spacing: .05em
+ letter-spacing: .05em;
+
+ .mat-simple-snackbar-action {
+ color: mat.get-contrast-color-from-palette($theming-material-components-success, 600);
+ }
}
.fail-snackbar {
- background: #850900;
- color: white;
+ background: mat.get-color-from-palette($theming-material-components-error, 600);
+ color: mat.get-contrast-color-from-palette($theming-material-components-error, 600);
font-weight: bold;
margin-bottom: 5px;
- letter-spacing: .05em
-}
+ letter-spacing: .05em;
-.warning-snackbar {
- background: rgb(255, 166, 0);
- color: white;
-}
-
-.success-snackbar .mat-simple-snackbar-action {
- color: black;
+ .mat-simple-snackbar-action {
+ color: mat.get-contrast-color-from-palette($theming-material-components-error, 600);
+ }
}
-.warning-snackbar .mat-simple-snackbar-action {
- color: black;
-}
+.warning-snackbar {
+ background: mat.get-color-from-palette($theming-material-components-warn, 600);
+ color: mat.get-contrast-color-from-palette($theming-material-components-warn, 600);
-.fail-snackbar .mat-simple-snackbar-action {
- color: black;
+ .mat-simple-snackbar-action {
+ color: mat.get-contrast-color-from-palette($theming-material-components-success, 600);
+ }
}
.chain-item {
background-color: var(--card-head-bg);
+
.ant-tabs-content {
min-height: 150px;
}
@@ -217,5 +223,110 @@ $theme-warn: mat.define-palette(mat.$yellow-palette);
justify-content: space-between;
}
-html, body { height: 100%; }
-body { margin: 0; font-family: Roboto, "Helvetica Neue", sans-serif; }
+html, body {
+ height: 100%;
+}
+
+body {
+ margin: 0;
+}
+
+.custom-class-tooltip {
+ transform: translateY(25px) !important;
+ background-color: black;
+ font-size: 12px;
+}
+
+mat-form-field {
+ &.flex-50 {
+ .mat-form-field-wrapper {
+ flex-basis: 50%;
+ }
+ }
+
+ &.flex-60 {
+ .mat-form-field-wrapper {
+ display: flex;
+ flex-basis: 60%;
+ }
+ }
+}
+
+table {
+ th.mat-header-cell {
+ color: mat.get-contrast-color-from-palette($theme-primary, 500);
+ }
+ .mat-row:hover {
+ background: mat.get-color-from-palette($theme-primary, 50);
+ }
+ .mat-header-row {
+ background: mat.get-color-from-palette($theme-primary);
+ }
+ .mat-header-cell:last-child {
+ border-right: none;
+ }
+}
+
+.flex-row-jc-ac {
+ display: flex;
+ flex-direction: row;
+ justify-content: center;
+ align-items: center;
+}
+
+.flex-row-jfs-ac {
+ display: flex;
+ flex-direction: row;
+ justify-content: flex-start;
+ align-items: center;
+}
+
+.flex-row-je {
+ display: flex;
+ flex-direction: row;
+ justify-content: flex-end;
+}
+
+.flex-row-jsb-ab {
+ display: flex;
+ flex-direction: row;
+ flex-wrap: nowrap;
+ justify-content: space-between;
+ align-items: baseline;
+}
+
+.flex-row-jsb-ac {
+ display: flex;
+ flex-direction: row;
+ flex-wrap: nowrap;
+ justify-content: space-between;
+ align-items: center;
+}
+
+.flex-row-jsb-ab-w {
+ display: flex;
+ flex-direction: row;
+ flex-wrap: wrap;
+ justify-content: space-between;
+ align-items: baseline;
+}
+
+.flex-cl-jsb-ab {
+ display: flex;
+ flex-direction: column;
+ flex-wrap: nowrap;
+ justify-content: space-between;
+ align-items: baseline;
+}
+
+.panel-class {
+ margin-top: 29px !important;
+}
+
+.mat-form-field-appearance-outline .mat-form-field-outline{
+ color: mat.get-color-from-palette($theme-primary);
+}
+
+.delete-icon {
+ color: mat.get-color-from-palette(mat.define-palette(mat.$red-palette), 300);
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/theme.scss b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/theme.scss
index 7b11c76e5..cc351b90e 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/theme.scss
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/frontend/parser-chains-client/src/theme.scss
@@ -8,13 +8,13 @@
$theming-material-components-primary: mat.define-palette($md-cyber-blue-primary);
$theming-material-components-accent: mat.define-palette($md-cyber-gray-secondary);
$theming-material-components-error: mat.define-palette(mat.$red-palette);
-$theming-material-components-warn: mat.define-palette(mat.$yellow-palette);
+$theming-material-components-warn: mat.define-palette(mat.$yellow-palette, 900, 50,50);
$theming-material-components-success: mat.define-palette(mat.$green-palette);
// Create the theme object (a Sass map containing all of the palettes).
$theming-material-components-theme: mat.define-light-theme(
$theming-material-components-primary,
- $theming-material-components-accent,
+ $theming-material-components-error,
$theming-material-components-warn
);
@@ -26,6 +26,9 @@ $typography: mat.define-typography-config(
$display-2: mat.define-typography-level(45px, $font-family: $heading-font-family),
$display-1: mat.define-typography-level(34px, $font-family: $heading-font-family),
$headline: mat.define-typography-level(24px, $font-family: $heading-font-family),
+ $body-1: mat.define-typography-level(16px, $font-family: $heading-font-family),
+ $body-2: mat.define-typography-level(14px, $font-family: $heading-font-family),
+ $caption: mat.define-typography-level(22px, $font-family: $heading-font-family),
$title: mat.define-typography-level(20px, $font-family: $heading-font-family),
);
@@ -34,6 +37,6 @@ $notifications-theme: (
default: #fff,
info: mat.get-color-from-palette(mat.define-palette(mat.$blue-palette), 400),
success: mat.get-color-from-palette(mat.define-palette(mat.$green-palette), 400),
- warning: mat.get-color-from-palette(mat.define-palette(mat.$yellow-palette), 400),
+ warning: mat.get-color-from-palette(mat.define-palette(mat.$red-palette), 400),
error: mat.get-color-from-palette(mat.define-palette(mat.$red-palette), 400),
);
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/pom.xml b/flink-cyber/metron-parser-chain/parser-chains-config-service/pom.xml
index 39fba15eb..424264f99 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/pom.xml
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/pom.xml
@@ -222,9 +222,8 @@
spring-boot-maven-plugin
${spring-boot.version}
-
-
- ZIP
+ com.cloudera.parserchains.queryservice.Application
+ JAR
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/StartupComponent.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/StartupComponent.java
index 4f3507569..2901675b8 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/StartupComponent.java
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/StartupComponent.java
@@ -46,5 +46,9 @@ public void run(String... args) throws Exception {
log.info("Creating the config dir: {}", appProperties.getConfigPath());
Files.createDirectories(Paths.get(appProperties.getConfigPath()));
log.info("Done creating the config dir");
+
+ log.info("Creating the pipeline dir: {}", appProperties.getPipelinesPath());
+ Files.createDirectories(Paths.get(appProperties.getPipelinesPath()));
+ log.info("Done creating the pipeline dir");
}
}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/common/exception/ValidationException.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/common/exception/ValidationException.java
new file mode 100644
index 000000000..41cb78f19
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/common/exception/ValidationException.java
@@ -0,0 +1,23 @@
+package com.cloudera.parserchains.queryservice.common.exception;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.web.bind.annotation.ResponseStatus;
+
+@ResponseStatus(value = HttpStatus.BAD_REQUEST)
+public class ValidationException extends RuntimeException {
+ public ValidationException() {
+ this(null,null);
+ }
+ public ValidationException(String message) {
+ this(message,null);
+ }
+ public ValidationException(Throwable cause) {
+ this(null,cause);
+ }
+ public ValidationException(String message, Throwable cause) {
+ super(message);
+ if (cause != null) {
+ this.initCause(cause);
+ }
+ }
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/common/utils/Utils.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/common/utils/Utils.java
new file mode 100644
index 000000000..675bb7393
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/common/utils/Utils.java
@@ -0,0 +1,120 @@
+package com.cloudera.parserchains.queryservice.common.utils;
+
+import org.apache.commons.lang.StringUtils;
+
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
+import java.time.format.DateTimeParseException;
+import java.time.temporal.ChronoField;
+import java.time.temporal.TemporalAccessor;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+
+
+public class Utils {
+
+ public static List DATE_FORMATS_CEF = new ArrayList() {
+ {
+ // as per CEF Spec
+ add(new SimpleDateFormat("MMM d HH:mm:ss.SSS Z"));
+ add(new SimpleDateFormat("MMM d HH:mm:ss.SSS z"));
+ add(new SimpleDateFormat("MMM d HH:mm:ss.SSS"));
+ add(new SimpleDateFormat("MMM d HH:mm:ss zzz"));
+ add(new SimpleDateFormat("MMM d HH:mm:ss"));
+ add(new SimpleDateFormat("MMM d yyyy HH:mm:ss.SSS Z"));
+ add(new SimpleDateFormat("MMM d yyyy HH:mm:ss.SSS z"));
+ add(new SimpleDateFormat("MMM d yyyy HH:mm:ss.SSS"));
+ add(new SimpleDateFormat("MMM d yyyy HH:mm:ss Z"));
+ add(new SimpleDateFormat("MMM d yyyy HH:mm:ss z"));
+ add(new SimpleDateFormat("MMM d yyyy HH:mm:ss"));
+ // found in the wild
+ add(new SimpleDateFormat("d MMMM yyyy HH:mm:ss"));
+ }
+ };
+
+ public static List DATE_FORMATS_SYSLOG = new ArrayList() {
+ {
+ // As specified in https://tools.ietf.org/html/rfc5424
+ add(new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSZ"));
+
+ // common format per rsyslog defaults e.g. Mar 21 14:05:02
+ add(new SimpleDateFormat("MMM dd HH:mm:ss"));
+ add(new SimpleDateFormat("MMM dd yyyy HH:mm:ss"));
+
+ // additional formats found in the wild
+ add(new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss"));
+ add(new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZ"));
+ add(new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS"));
+
+ }
+ };
+
+ public static List DATE_FORMATS = new ArrayList() {
+ {
+ add(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSSZ"));
+
+ // common format per rsyslog defaults e.g. Mar 21 14:05:02
+ add(new SimpleDateFormat("MMM dd HH:mm:ss"));
+ add(new SimpleDateFormat("MMM dd yyyy HH:mm:ss"));
+
+ // additional formats found in the wild
+ add(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));
+ add(new SimpleDateFormat("yyyy-MM-dd HH:mm:ssZ"));
+ add(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"));
+
+ }
+ };
+
+
+ /**
+ * Parse the data according to a sequence of possible parse patterns
+ * If the given date is entirely numeric, it is assumed to be a unix
+ * timestamp.
+ * If the year is not specified in the date string, use the current year.
+ * Assume that any date more than 4 days in the future is in the past as per
+ * SyslogUtils
+ *
+ * @param candidate
+ * The possible date.
+ * @param validPatterns
+ * A list of SimpleDateFormat instances to try parsing with.
+ * @return A java.util.Date based on the parse result
+ */
+ public static Long parseData(String candidate, List validPatterns) {
+ if (StringUtils.isNumeric(candidate)) {
+ return Long.parseLong(candidate);
+ } else {
+ for (SimpleDateFormat pattern : validPatterns) {
+ try {
+ DateTimeFormatterBuilder formatterBuilder = new DateTimeFormatterBuilder()
+ .appendPattern(pattern.toPattern());
+ DateTimeFormatter formatter = formatterBuilder.toFormatter();
+ ZonedDateTime parsedValue = parseDateTimeWithDefaultTimezone(candidate, formatter);
+ return parsedValue.toInstant().toEpochMilli();
+ } catch (DateTimeParseException ignored) {
+ // Continue to the next pattern
+ }
+ }
+ return null;
+ }
+ }
+
+ private static ZonedDateTime parseDateTimeWithDefaultTimezone(String candidate, DateTimeFormatter formatter) {
+ TemporalAccessor temporalAccessor = formatter.parseBest(candidate, ZonedDateTime::from, LocalDateTime::from);
+ return temporalAccessor instanceof ZonedDateTime
+ ? ((ZonedDateTime) temporalAccessor)
+ : ((LocalDateTime) temporalAccessor).atZone(ZoneId.systemDefault());
+ }
+
+ public static int compareLongs(Long a, Long b) {
+ Comparator comparator = Comparator.nullsFirst(Long::compareTo);
+ return comparator.compare(a, b);
+ }
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/config/kafka/KafkaConfig.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/config/kafka/KafkaConfig.java
index b63159109..9e9c6aa2d 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/config/kafka/KafkaConfig.java
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/config/kafka/KafkaConfig.java
@@ -1,5 +1,8 @@
package com.cloudera.parserchains.queryservice.config.kafka;
+import com.cloudera.parserchains.queryservice.service.KafkaService;
+import com.cloudera.parserchains.queryservice.service.KafkaServiceInterface;
+import com.cloudera.parserchains.queryservice.service.MockKafkaService;
import com.cloudera.service.common.config.kafka.ClouderaKafkaProperties;
import com.cloudera.service.common.request.RequestBody;
import com.cloudera.service.common.response.ResponseBody;
@@ -8,6 +11,7 @@
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
@@ -57,6 +61,7 @@ public ClouderaKafkaProperties kafkaProperties() {
@Bean(name = "kafka-external-cluster-map")
@Scope(ConfigurableBeanFactory.SCOPE_SINGLETON)
@ConfigurationProperties("spring.kafka.external-clusters")
+ @ConditionalOnProperty(name = "spring.kafka.mock", matchIfMissing = true, havingValue = "false")
public Map replyKafkaPropertiesMap() {
return new HashMap<>();
}
@@ -71,6 +76,7 @@ public Map replyKafkaPropertiesMap() {
*/
@Bean(name = "kafkaTemplatePool")
@Scope(ConfigurableBeanFactory.SCOPE_SINGLETON)
+ @ConditionalOnProperty(name = "spring.kafka.mock", matchIfMissing = true, havingValue = "false")
public Map> kafkaTemplatePool(
@Qualifier("kafka-external-cluster-map") Map replyKafkaPropertiesMap) {
final Map> templatePool = new HashMap<>();
@@ -93,9 +99,23 @@ public Map kafkaClustersSet(@Qualifier("kafka-external-cluster-map") Map replyKafkaPropertiesMap) {
return Collections.unmodifiableSet(replyKafkaPropertiesMap.keySet());
}
+ @Bean
+ @ConditionalOnProperty(name = "spring.kafka.mock")
+ public KafkaServiceInterface mockKafkaService() {
+ return new MockKafkaService();
+ }
+
+ @Bean
+ @ConditionalOnProperty(name = "spring.kafka.mock", matchIfMissing = true, havingValue = "false")
+ public KafkaServiceInterface kafkaService(@Qualifier("kafkaTemplatePool") Map> kafkaTemplatePool,
+ @Value("${kafka.reply.future.timeout:45}") Long replyFutureTimeout,
+ @Value("${kafka.reply.timeout:45}") Long kafkaTemplateTimeout) {
+ return new KafkaService(kafkaTemplatePool,replyFutureTimeout,kafkaTemplateTimeout);
+ }
private ProducerFactory producerFactory(ClouderaKafkaProperties kafkaProperties) {
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ChainController.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ChainController.java
index e990cb19f..d93dace30 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ChainController.java
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ChainController.java
@@ -75,44 +75,32 @@ public class ChainController {
* The maximum number of sample text values that can be used to test a parser chain.
*/
static final int MAX_SAMPLES_PER_TEST = 200;
-
private final ChainPersistenceService chainPersistenceService;
-
private final ChainBuilderService chainBuilderService;
-
private final ChainExecutorService chainExecutorService;
-
private final PipelineService pipelineService;
-
private final IndexingService indexingService;
-
private final AppProperties appProperties;
-
- @Operation(summary = "Retrieves all available parser chains.",
- responses = {
- @ApiResponse(responseCode = "200", description = "A list of all parser chains.", content = @Content(
- mediaType = "application/json",
- array = @ArraySchema(schema = @Schema(implementation = ParserChainSummary.class))))
- })
+ @Operation(summary = "Retrieves all available parser chains.")
+ @ApiResponse(responseCode = "200", description = "A list of all parser chains.", content = @Content(
+ mediaType = "application/json",
+ array = @ArraySchema(schema = @Schema(implementation = ParserChainSummary.class))))
@GetMapping(value = API_CHAINS)
public ResponseEntity> findAll(
@Parameter(name = "pipelineName", description = "The pipeline to execute request in.")
@RequestParam(name = "pipelineName", required = false) String pipelineName
) throws IOException {
String configPath = getConfigPath(pipelineName);
-
List configs = chainPersistenceService.findAll(Paths.get(configPath));
return ResponseEntity.ok(configs);
}
- @Operation(summary = "Creates a new parser chain.",
- responses = {
- @ApiResponse(responseCode = "200", description = "The parser chain was created.", content = @Content(
- mediaType = "application/json",
- schema = @Schema(implementation = ParserChainSchema.class))),
- @ApiResponse(responseCode = "404", description = "Unable to create a new parser chain.")
- })
+ @Operation(summary = "Creates a new parser chain.")
+ @ApiResponse(responseCode = "200", description = "The parser chain was created.", content = @Content(
+ mediaType = "application/json",
+ schema = @Schema(implementation = ParserChainSchema.class)))
+ @ApiResponse(responseCode = "404", description = "Unable to create a new parser chain.")
@PostMapping(value = API_CHAINS)
public ResponseEntity create(
@Parameter(name = "pipelineName", description = "The pipeline to execute request in.")
@@ -120,25 +108,19 @@ public ResponseEntity create(
@Parameter(name = "parserChain", description = "The parser chain to create.", required = true)
@RequestBody ParserChainSchema chain) throws IOException {
String configPath = getConfigPath(pipelineName);
-
ParserChainSchema createdChain = chainPersistenceService.create(chain, Paths.get(configPath));
- if (null == createdChain) {
+ if (createdChain == null) {
return ResponseEntity.notFound().build();
} else {
- return ResponseEntity
- .created(URI.create(API_CHAINS_READ_URL.replace("{id}", createdChain.getId())))
- .body(createdChain);
+ return ResponseEntity.created(URI.create(API_CHAINS_READ_URL.replace("{id}", createdChain.getId()))).body(createdChain);
}
}
-
- @Operation(summary = "Retrieves an existing parser chain.",
- responses = {
- @ApiResponse(responseCode = "200", description = "The parser chain with the given ID.", content = @Content(
- mediaType = "application/json",
- schema = @Schema(implementation = ParserChainSchema.class))),
- @ApiResponse(responseCode = "404", description = "The parser chain does not exist.")
- })
+ @Operation(summary = "Retrieves an existing parser chain.")
+ @ApiResponse(responseCode = "200", description = "The parser chain with the given ID.", content = @Content(
+ mediaType = "application/json",
+ schema = @Schema(implementation = ParserChainSchema.class)))
+ @ApiResponse(responseCode = "404", description = "The parser chain does not exist.")
@GetMapping(value = API_CHAINS + "/{id}")
public ResponseEntity read(
@Parameter(name = "pipelineName", description = "The pipeline to execute request in.")
@@ -146,21 +128,17 @@ public ResponseEntity read(
@Parameter(name = "id", description = "The ID of the parser chain to retrieve.", required = true)
@PathVariable String id) throws IOException {
String configPath = getConfigPath(pipelineName);
-
ParserChainSchema chain = chainPersistenceService.read(id, Paths.get(configPath));
- if (null == chain) {
+ if (chain == null) {
return ResponseEntity.notFound().build();
} else {
return ResponseEntity.ok(chain);
}
}
-
- @Operation(summary = "Updates an existing parser chain.",
- responses = {
- @ApiResponse(responseCode = "204", description = "The parser chain was updated."),
- @ApiResponse(responseCode = "404", description = "The parser chain does not exist.")
- })
+ @Operation(summary = "Updates an existing parser chain.")
+ @ApiResponse(responseCode = "204", description = "The parser chain was updated.")
+ @ApiResponse(responseCode = "404", description = "The parser chain does not exist.")
@PutMapping(value = API_CHAINS + "/{id}")
public ResponseEntity update(
@Parameter(name = "pipelineName", description = "The pipeline to execute request in.")
@@ -177,17 +155,14 @@ public ResponseEntity update(
} else {
return ResponseEntity.noContent().build();
}
- // TODO: fix exception handling
} catch (IOException ioe) {
throw new RuntimeException("Unable to update configuration with id=" + id);
}
}
- @Operation(summary = "Deletes a parser chain."
- , responses = {
- @ApiResponse(responseCode = "204", description = "The parser chain was deleted."),
- @ApiResponse(responseCode = "404", description = "The parser chain does not exist.")
- })
+ @Operation(summary = "Deletes a parser chain.")
+ @ApiResponse(responseCode = "204", description = "The parser chain was deleted.")
+ @ApiResponse(responseCode = "404", description = "The parser chain does not exist.")
@DeleteMapping(value = API_CHAINS + "/{id}")
public ResponseEntity delete(
@Parameter(name = "pipelineName", description = "The pipeline to execute request in.")
@@ -195,7 +170,6 @@ public ResponseEntity delete(
@Parameter(name = "id", description = "The ID of the parser chain to delete.", required = true)
@PathVariable String id) throws IOException {
String configPath = getConfigPath(pipelineName);
-
if (chainPersistenceService.delete(id, Paths.get(configPath))) {
return ResponseEntity.noContent().build();
} else {
@@ -247,12 +221,10 @@ public ResponseEntity saveMappingsToPath(
}
}
- @Operation(summary = "Executes a parser chain to parse sample data.",
- responses = {
- @ApiResponse(responseCode = "200", description = "The result of parsing the message.", content = @Content(
- mediaType = "application/json",
- schema = @Schema(implementation = ChainTestResponse.class)))
- })
+ @Operation(summary = "Executes a parser chain to parse sample data.")
+ @ApiResponse(responseCode = "200", description = "The result of parsing the message.", content = @Content(
+ mediaType = "application/json",
+ schema = @Schema(implementation = ChainTestResponse.class)))
@PostMapping(value = API_PARSER_TEST)
public ResponseEntity test(
@Parameter(name = "pipelineName", description = "The pipeline to execute request in.")
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ClusterController.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ClusterController.java
index 05b1099f8..afb97ba16 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ClusterController.java
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ClusterController.java
@@ -4,6 +4,7 @@
import com.cloudera.parserchains.queryservice.common.exception.FailedAllClusterReponseException;
import com.cloudera.parserchains.queryservice.common.exception.FailedClusterReponseException;
import com.cloudera.parserchains.queryservice.service.ClusterService;
+import com.cloudera.service.common.response.Pipeline;
import com.cloudera.service.common.response.ResponseBody;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
@@ -12,9 +13,14 @@
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestPart;
import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.multipart.MultipartFile;
+import java.io.IOException;
import java.util.List;
/**
@@ -48,4 +54,55 @@ public ResponseBody getClusterService(
@PathVariable("id") String clusterId) throws FailedClusterReponseException {
return clusterService.getClusterInfo(clusterId);
}
+
+ @Operation(description = "Retrieves information about all pipelines on all services.")
+ @ApiResponses(value = {
+ @ApiResponse(responseCode = "200", description = "A list of all pipelines.")
+ })
+ @GetMapping(value = "/pipelines")
+ public List getAllPipelines() throws FailedAllClusterReponseException {
+ return clusterService.getAllPipelines();
+ }
+
+ @Operation(description = "Retrieves information about all pipelines on all services.")
+ @ApiResponses(value = {
+ @ApiResponse(responseCode = "200", description = "A list of all pipelines.")
+ })
+ @PostMapping(value = "{clusterId}/pipelines")
+ public ResponseBody createEmptyPipeline(
+ @RequestBody com.cloudera.service.common.request.RequestBody body,
+ @Parameter(name = "clusterId", description = "The ID of the cluster to update config on.", required = true)
+ @PathVariable("clusterId") String clusterId
+ ) throws FailedClusterReponseException {
+ return clusterService.createEmptyPipeline(clusterId, body);
+ }
+
+ @Operation(description = "Retrieves information about all pipelines on all services.")
+ @ApiResponses(value = {
+ @ApiResponse(responseCode = "200", description = "A list of all pipelines.")
+ })
+ @PostMapping(value = "{clusterId}/pipelines/{name}/start")
+ public ResponseBody startPipeline(
+ @Parameter(name = "name", description = "The pipeline name to create empty pipeline.", required = true)
+ @PathVariable("name") String name,
+ @Parameter(name = "clusterId", description = "The ID of the cluster to update config on.", required = true)
+ @PathVariable("clusterId") String clusterId,
+ @RequestPart("payload") MultipartFile payload,
+ @RequestPart("body") com.cloudera.service.common.request.RequestBody body
+ ) throws IOException, FailedClusterReponseException {
+ return clusterService.startPipelineJob(clusterId, name, body.getBranch(), body.getProfileName(), body.getJobs(), payload.getBytes());
+ }
+
+ @Operation(description = "Retrieves information about a pipeline on the cluster with specified id.")
+ @ApiResponses(value = {
+ @ApiResponse(responseCode = "200", description = "A response with cluster information."),
+ @ApiResponse(responseCode = "404", description = "The cluster does not exist.")
+
+ })
+ @GetMapping(value = "/{id}/pipelines")
+ public List getClusterPipelines(
+ @Parameter(name = "id", description = "The ID of the cluster to retrieve.", required = true)
+ @PathVariable("id") String clusterId) throws FailedClusterReponseException {
+ return clusterService.getClusterPipelines(clusterService.getClusterInfo(clusterId));
+ }
}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/JobController.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/JobController.java
index b9d7c10eb..5ec57ab75 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/JobController.java
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/JobController.java
@@ -72,6 +72,4 @@ public ResponseBody updateJobConfig(@Parameter(name = "clusterId", description =
.build();
return jobService.makeRequest(clusterId, requestBody, JobActions.Constants.UPDATE_CONFIG_VALUE);
}
-
-
}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ParserController.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ParserController.java
index 6fa6d0ad4..6a2294627 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ParserController.java
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ParserController.java
@@ -23,7 +23,6 @@
import io.swagger.v3.oas.annotations.media.Schema;
import io.swagger.v3.oas.annotations.responses.ApiResponse;
import lombok.RequiredArgsConstructor;
-import org.apache.commons.collections4.MapUtils;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
@@ -45,17 +44,12 @@
@RequestMapping(value = PARSER_CONFIG_BASE_URL)
public class ParserController {
-
private final ParserDiscoveryService parserDiscoveryService;
-
-
- @Operation(summary = "Retrieves all available parsers.",
- responses = {
- @ApiResponse(responseCode = "200", description = "A list of all parser types.", content = @Content(
- mediaType = "application/json",
- array = @ArraySchema(schema = @Schema(implementation = ParserSummary.class))))
- })
+ @Operation(summary = "Retrieves all available parsers.")
+ @ApiResponse(responseCode = "200", description = "A list of all parser types.", content = @Content(
+ mediaType = "application/json",
+ array = @ArraySchema(schema = @Schema(implementation = ParserSummary.class))))
@GetMapping(value = API_PARSER_TYPES)
public ResponseEntity> findAll() throws IOException {
List types = parserDiscoveryService.findAll();
@@ -63,11 +57,10 @@ public ResponseEntity> findAll() throws IOException {
}
- @Operation(summary = "Describes the configuration parameters for all available parsers.",
- responses = {
- @ApiResponse(responseCode = "200", description = "A map of parser types and their associated configuration parameters.", content = @Content(
- mediaType = "application/json")),
- @ApiResponse(responseCode = "404", description = "Unable to retrieve.")})
+ @Operation(summary = "Describes the configuration parameters for all available parsers.")
+ @ApiResponse(responseCode = "200", description = "A map of parser types and their associated configuration parameters.", content = @Content(
+ mediaType = "application/json"))
+ @ApiResponse(responseCode = "404", description = "Unable to retrieve.")
@GetMapping(value = API_PARSER_FORM_CONFIG)
public ResponseEntity> describeAll() {
Map configs = parserDiscoveryService.describeAll();
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ParserSampleController.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ParserSampleController.java
index 86dbac76b..480a4cef7 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ParserSampleController.java
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/ParserSampleController.java
@@ -18,27 +18,17 @@
package com.cloudera.parserchains.queryservice.controller;
-import static com.cloudera.parserchains.queryservice.common.ApplicationConstants.API_PARSER_TEST_SAMPLES;
-import static com.cloudera.parserchains.queryservice.common.ApplicationConstants.PARSER_CONFIG_BASE_URL;
-
import com.cloudera.parserchains.queryservice.config.AppProperties;
import com.cloudera.parserchains.queryservice.model.describe.SampleFolderDescriptor;
import com.cloudera.parserchains.queryservice.model.sample.ParserSample;
import com.cloudera.parserchains.queryservice.service.ParserSampleService;
-
-import java.io.IOException;
-import java.net.URI;
-import java.util.List;
-import lombok.RequiredArgsConstructor;
-
-
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
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.springframework.beans.factory.annotation.Autowired;
+import lombok.RequiredArgsConstructor;
import org.springframework.http.ResponseEntity;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.PathVariable;
@@ -48,6 +38,13 @@
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
+import java.io.IOException;
+import java.net.URI;
+import java.util.List;
+
+import static com.cloudera.parserchains.queryservice.common.ApplicationConstants.API_PARSER_TEST_SAMPLES;
+import static com.cloudera.parserchains.queryservice.common.ApplicationConstants.PARSER_CONFIG_BASE_URL;
+
/**
* The controller responsible for operations on parsers.
*/
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/PipelineController.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/PipelineController.java
index d8eef055d..32537c2f7 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/PipelineController.java
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/controller/PipelineController.java
@@ -15,6 +15,7 @@
import com.cloudera.parserchains.queryservice.model.exec.PipelineResult;
import com.cloudera.parserchains.queryservice.service.PipelineService;
import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.responses.ApiResponse;
import io.swagger.v3.oas.annotations.responses.ApiResponses;
import lombok.RequiredArgsConstructor;
@@ -63,7 +64,8 @@ public ResponseEntity> findAll() throws IOException {
@ApiResponse(responseCode = "200", description = "A new list of all pipelines.")
})
@PostMapping("/{pipelineName}")
- public ResponseEntity> createPipeline(@PathVariable String pipelineName) throws IOException {
+ public ResponseEntity> createPipeline(
+ @Parameter(description = "The name of the pipeline to create") @PathVariable String pipelineName) throws IOException {
PipelineResult newPipeline = pipelineService.createPipeline(pipelineName);
if (newPipeline != null) {
return findAll();
@@ -76,8 +78,9 @@ public ResponseEntity> createPipeline(@PathVariable String pipelineN
@ApiResponse(responseCode = "200", description = "A new list of all pipelines.")
})
@PutMapping("/{pipelineName}")
- public ResponseEntity> renamePipeline(@PathVariable String pipelineName,
- @RequestParam String newName) throws IOException {
+ public ResponseEntity> renamePipeline(
+ @Parameter(description = "The current name of the pipeline to be renamed") @PathVariable String pipelineName,
+ @Parameter(description = "The new name for the pipeline") @RequestParam String newName) throws IOException {
PipelineResult updatedPipeline = pipelineService.renamePipeline(pipelineName, newName);
if (updatedPipeline != null) {
return findAll();
@@ -90,12 +93,12 @@ public ResponseEntity> renamePipeline(@PathVariable String pipelineN
@ApiResponse(responseCode = "200", description = "A new list of all pipelines.")
})
@DeleteMapping("/{pipelineName}")
- public ResponseEntity> deletePipeline(@PathVariable String pipelineName) throws IOException {
+ public ResponseEntity> deletePipeline(
+ @Parameter(description = "The name of the pipeline to be deleted") @PathVariable String pipelineName) throws IOException {
boolean pipelineDeleted = pipelineService.deletePipeline(pipelineName);
if (pipelineDeleted) {
return findAll();
}
return ResponseEntity.badRequest().build();
}
-
}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/ClusterService.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/ClusterService.java
index c1e3c3d30..721f801d0 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/ClusterService.java
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/ClusterService.java
@@ -2,23 +2,30 @@
import com.cloudera.parserchains.queryservice.common.exception.FailedAllClusterReponseException;
import com.cloudera.parserchains.queryservice.common.exception.FailedClusterReponseException;
+import com.cloudera.parserchains.queryservice.common.utils.Utils;
import com.cloudera.service.common.request.RequestBody;
import com.cloudera.service.common.request.RequestType;
+import com.cloudera.service.common.response.Job;
+import com.cloudera.service.common.response.Pipeline;
import com.cloudera.service.common.response.ResponseBody;
import com.cloudera.service.common.response.ResponseType;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.stereotype.Service;
+import java.util.ArrayList;
+import java.util.Base64;
import java.util.List;
+import java.util.Optional;
import java.util.stream.Collectors;
@Slf4j
@Service
@RequiredArgsConstructor
public class ClusterService {
- private final KafkaService kafkaService;
+ private final KafkaServiceInterface kafkaService;
public List getAllClusterInfo() throws FailedAllClusterReponseException {
List> response = kafkaService.sendWithReply(RequestType.GET_ALL_CLUSTERS_SERVICE_REQUEST, RequestBody.builder().build());
@@ -32,6 +39,60 @@ public List getAllClusterInfo() throws FailedAllClusterReponseExce
return response.stream().map(Pair::getValue).collect(Collectors.toList());
}
+ public List getAllPipelines() throws FailedAllClusterReponseException {
+ return getAllClusterInfo().stream()
+ .flatMap(responseBody -> getClusterPipelines(responseBody).stream())
+ .collect(Collectors.toList());
+ }
+
+ public List getClusterPipelines(ResponseBody responseBody) {
+ return responseBody.getJobs().stream().collect(ArrayList::new,
+ (acc, job) -> {
+ Optional duplicatePipeline = acc.stream().filter(pipeline -> StringUtils.equalsIgnoreCase(pipeline.getName(), job.getJobPipeline())).findFirst();
+ if (duplicatePipeline.isPresent()) {
+ Pipeline pipeline = duplicatePipeline.get();
+ if (isNewJobTimeLater(job, pipeline)) {
+ pipeline.setDate(job.getStartTime());
+ }
+ } else {
+ acc.add(Pipeline.builder()
+ .name(job.getJobPipeline())
+ .clusterName(responseBody.getClusterMeta().getName())
+ .date(job.getStartTime())
+ .userName(job.getUser())
+ .build());
+ }
+ }, ArrayList::addAll);
+ }
+
+ public ResponseBody createEmptyPipeline(String clusterId, RequestBody body) throws FailedClusterReponseException {
+ RequestBody requestBody = RequestBody.builder()
+ .pipelineName(body.getPipelineName())
+ .branch(body.getBranch())
+ .build();
+ Pair response = kafkaService.sendWithReply(RequestType.CREATE_EMPTY_PIPELINE, clusterId, requestBody);
+ if (response.getKey() != ResponseType.CREATE_EMPTY_PIPELINE_RESPONSE) {
+ throw new FailedClusterReponseException(response.getValue());
+ }
+ return response.getValue();
+ }
+
+ public ResponseBody startPipelineJob(String clusterId, String pipeline, String branch, String profileName, List jobs, byte[] payload) throws FailedClusterReponseException {
+ Pair response = kafkaService.sendWithReply(RequestType.START_ARCHIVE_PIPELINE, clusterId, RequestBody
+ .builder()
+ .payload(Base64.getEncoder().encode(payload))
+ .branch(branch)
+ .profileName(profileName)
+ .jobs(jobs)
+ .pipelineName(pipeline)
+ .build());
+ if (response.getKey() != ResponseType.CREATE_EMPTY_PIPELINE_RESPONSE) {
+ throw new FailedClusterReponseException(response.getValue());
+ }
+ return response.getValue();
+ }
+
+
public ResponseBody getClusterInfo(String clusterId) throws FailedClusterReponseException {
return getClusterInfo(clusterId, RequestBody.builder().build());
}
@@ -43,4 +104,9 @@ public ResponseBody getClusterInfo(String clusterId, RequestBody body) throws Fa
}
return response.getValue();
}
+
+ private static boolean isNewJobTimeLater(Job job, Pipeline pipeline) {
+ return Utils.compareLongs(Utils.parseData(job.getStartTime(), Utils.DATE_FORMATS),
+ (Utils.parseData(pipeline.getDate(), Utils.DATE_FORMATS))) == 1;
+ }
}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/JobService.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/JobService.java
index e194d4523..4e862395f 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/JobService.java
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/JobService.java
@@ -20,7 +20,7 @@
@Service
@RequiredArgsConstructor
public class JobService {
- private final KafkaService kafkaService;
+ private final KafkaServiceInterface kafkaService;
private final ClusterService clusterService;
public ResponseBody makeRequest(String clusterId, RequestBody body, String actionString) throws FailedClusterReponseException {
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/KafkaService.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/KafkaService.java
index 2e20f6e8b..c7b2c5d65 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/KafkaService.java
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/KafkaService.java
@@ -31,22 +31,12 @@
import java.util.stream.Collectors;
@Slf4j
-@Service
-public class KafkaService {
+@RequiredArgsConstructor
+public class KafkaService implements KafkaServiceInterface {
private final Map> kafkaTemplatePool;
private final Long replyFutureTimeout;
private final Long kafkaTemplateTimeout;
-
- public KafkaService(@Qualifier("kafkaTemplatePool") Map> kafkaTemplatePool,
- @Value("${kafka.reply.future.timeout:45}") Long replyFutureTimeout,
- @Value("${kafka.reply.timeout:45}") Long kafkaTemplateTimeout) {
- this.kafkaTemplatePool = kafkaTemplatePool;
- this.replyFutureTimeout = replyFutureTimeout;
- this.kafkaTemplateTimeout = kafkaTemplateTimeout;
- }
-
-
public Pair sendWithReply(RequestType requestType, String clusterId, RequestBody body) {
final ClouderaReplyingKafkaTemplate kafkaTemplate = kafkaTemplatePool.get(clusterId);
if (kafkaTemplate == null) {
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/KafkaServiceInterface.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/KafkaServiceInterface.java
new file mode 100644
index 000000000..4914b7f1b
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/KafkaServiceInterface.java
@@ -0,0 +1,15 @@
+package com.cloudera.parserchains.queryservice.service;
+
+import com.cloudera.service.common.request.RequestBody;
+import com.cloudera.service.common.request.RequestType;
+import com.cloudera.service.common.response.ResponseBody;
+import com.cloudera.service.common.response.ResponseType;
+import org.apache.commons.lang3.tuple.Pair;
+
+import java.util.List;
+
+public interface KafkaServiceInterface {
+ Pair sendWithReply(RequestType requestType, String clusterId, RequestBody body);
+
+ List> sendWithReply(RequestType requestType, RequestBody body);
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/MockKafkaService.java b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/MockKafkaService.java
new file mode 100644
index 000000000..19ba7cb86
--- /dev/null
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/java/com/cloudera/parserchains/queryservice/service/MockKafkaService.java
@@ -0,0 +1,120 @@
+package com.cloudera.parserchains.queryservice.service;
+
+import com.cloudera.service.common.request.RequestBody;
+import com.cloudera.service.common.request.RequestType;
+import com.cloudera.service.common.response.ClusterMeta;
+import com.cloudera.service.common.response.Job;
+import com.cloudera.service.common.response.ResponseBody;
+import com.cloudera.service.common.response.ResponseType;
+import com.google.common.collect.Lists;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class MockKafkaService implements KafkaServiceInterface {
+ private final ClusterMeta cluster1 = ClusterMeta.builder()
+ .clusterId("1")
+ .name("Cluster1")
+ .flinkVersion("3.2.1")
+ .version("1.0.1")
+ .clusterStatus("online")
+ .build();
+ private final ClusterMeta cluster2 = ClusterMeta.builder()
+ .clusterId("2")
+ .name("Cluster2")
+ .flinkVersion("3.4.1")
+ .version("1.0.5")
+ .clusterStatus("online")
+ .build();
+ private final ClusterMeta cluster3 = ClusterMeta.builder()
+ .clusterId("3")
+ .name("Cluster3")
+ .flinkVersion("3.1.1")
+ .version("1.0.4")
+ .clusterStatus("offline")
+ .build();
+
+ private final Job job1 = Job.builder()
+ .jobId(JobID.fromHexString("fd72014d4c864993a2e5a9287b4a9c5d"))
+ .jobName("job1")
+ .jobIdString("job1-id-string")
+ .jobBranch("main")
+ .jobState(JobStatus.RUNNING)
+ .jobFullName("main-job1")
+ .jobPipeline("cyb-33")
+ .startTime("2024-01-11 10:33:55")
+ .jobType(Job.JobType.PARSER)
+ .user("Cybersec")
+ .build();
+ private final Job job2 = Job.builder()
+ .jobId(JobID.fromHexString("ad72014d4c864993a2e5a9287b4a9c5d"))
+ .jobName("job2")
+ .jobIdString("job2-id-string")
+ .jobBranch("main")
+ .jobState(JobStatus.RUNNING)
+ .jobFullName("main-job2")
+ .jobPipeline("cyb-33")
+ .startTime("2024-01-14 14:33:55")
+ .jobType(Job.JobType.PARSER)
+ .user("Cybersec")
+ .build();
+ private final Job job3 = Job.builder()
+ .jobId(JobID.fromHexString("cd72014d4c864993a2e5a9287b4a9c5d"))
+ .jobName("job3")
+ .jobIdString("job3-id-string")
+ .jobBranch("main")
+ .jobState(JobStatus.RUNNING)
+ .jobFullName("main-job3")
+ .jobPipeline("cyb-43")
+ .startTime("2024-01-11 11:33:55")
+ .jobType(Job.JobType.PARSER)
+ .user("Cybersec")
+ .build();
+ private final Job job4 = Job.builder()
+ .jobId(JobID.fromHexString("bd72014d4c864993a2e5a9287b4a9c5d"))
+ .jobName("job4")
+ .jobIdString("job1-id-string")
+ .jobBranch("main")
+ .jobState(JobStatus.RUNNING)
+ .jobFullName("main-job4")
+ .jobPipeline("cyb-43")
+ .startTime("2024-01-15 15:33:55")
+ .jobType(Job.JobType.PARSER)
+ .user("Cybersec")
+ .build();
+
+ private final ResponseBody responseFromCluster1 = ResponseBody.builder()
+ .jobs(Lists.newArrayList(job1, job2, job4))
+ .clusterMeta(cluster1)
+ .build();
+ private final ResponseBody responseFromCluster2 = ResponseBody.builder()
+ .jobs(Lists.newArrayList(job1))
+ .clusterMeta(cluster2)
+ .build();
+ private final ResponseBody responseFromCluster3 = ResponseBody.builder()
+ .jobs(Lists.newArrayList(job3))
+ .clusterMeta(cluster3)
+ .build();
+ private final List responses = Lists.newArrayList(responseFromCluster1, responseFromCluster2, responseFromCluster3);
+
+ @Override
+ public Pair sendWithReply(RequestType requestType, String clusterId, RequestBody body) {
+ return responses.stream()
+ .filter(response -> StringUtils.equalsIgnoreCase(response.getClusterMeta().getName(), clusterId))
+ .map(val -> Pair.of(ResponseType.GET_CLUSTER_SERVICE_RESPONSE, val))
+ .findFirst()
+ .orElse(Pair.of(ResponseType.GET_CLUSTER_SERVICE_RESPONSE, responseFromCluster1));
+ }
+
+ @Override
+ public List> sendWithReply(RequestType requestType, RequestBody body) {
+ return responses.stream()
+ .map(val -> Pair.of(ResponseType.GET_ALL_CLUSTERS_SERVICE_RESPONSE, val))
+ .collect(Collectors.toList());
+ }
+}
diff --git a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/resources/application.yml b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/resources/application.yml
index fef60e446..095757ea6 100644
--- a/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/resources/application.yml
+++ b/flink-cyber/metron-parser-chain/parser-chains-config-service/src/main/resources/application.yml
@@ -6,6 +6,7 @@ server:
tomcat:
accesslog:
enabled: true
+ port: 3000
# Enable/disable and expose/hide web enpoints for Actuator admin services
management.endpoints.web:
@@ -19,7 +20,8 @@ logging:
apache:
tomcat: INFO
catalina: INFO
-
+pipelines:
+ path: "../pipelines"
# Swagger
springdoc:
@@ -27,11 +29,12 @@ springdoc:
path: /api-docs
swagger-ui:
path: /swagger-ui
-
-
# Kafka
spring:
+ profiles:
+ active: "kafka"
kafka:
+ mock: true
bootstrap-servers: localhost:9092
reply-topic: reply-topic
request-topic: request-topic
|