Stream logs & Task Execution Logs

Stream Summary: add log section when the stream is running
Task Execution: add log section after execution done
Resolves #1230 #1231

Author: oodamien

ui/src/app/streams/stream/styles.scss

@@ -45,3 +45,14 @@ table.table {
     width: auto;
   }
 }
+
+.log-toolbar {
+  padding-bottom: 10px;
+  .dropdown {
+    display: inline;
+    margin-right: 4px;
+    ul li a {
+      cursor: pointer;
+    }
+  }
+}

ui/src/app/streams/stream/summary/stream-summary.component.html

@@ -100,6 +100,36 @@
           </tbody>
         </table>
       </div>
+
+      <div class="row stream-summary-row">
+        <div class="col-md-3">
+          <strong>Logs:</strong>
+        </div>
+        <div class="col-md-21">
+          <div class="log-toolbar">
+            <div dropdown class="dropdown">
+              <a class="btn btn-default btn-dropdown" dropdownToggle>
+                <strong>{{ stream.logs[logSelectedIndex].id }}</strong>
+                <span class="caret"></span>
+              </a>
+              <ul class="dropdown-menu" *dropdownMenu="">
+                <ng-template ngFor let-runtimeApp [ngForOf]="stream.runtimes" let-i="index">
+                  <li [class.active]="stream.logs[logSelectedIndex].id === runtimeApp.id">
+                    <a (click)="changeLog(i)">{{ runtimeApp.id }}</a>
+                  </li>
+                </ng-template>
+              </ul>
+            </div>
+            <button type="button" class="btn btn-primary" (click)="refresh()">
+              Refresh
+            </button>
+          </div>
+          <div class="pui-log" *ngIf="stream.logs[logSelectedIndex].log">
+            <pre><code>{{ stream.logs[logSelectedIndex].log }}</code></pre>
+          </div>
+        </div>
+      </div>
+
     </div>
 
   </div>

ui/src/app/streams/stream/summary/stream-summary.component.ts

@@ -31,6 +31,11 @@ export class StreamSummaryComponent implements OnInit {
    */
   modal: BsModalRef;
 
+  /**
+   * Log Selected
+   */
+  logSelectedIndex = 0;
+
   /**
    * Constructor
    *
@@ -77,12 +82,43 @@ export class StreamSummaryComponent implements OnInit {
           }
         }
       ))
+      .pipe(mergeMap(
+        (val: any) => {
+          if (val.runtimes.length > 0) {
+            return this.streamsService.getLogs(val.streamDefinition)
+              .pipe(map((logs) => {
+                return {
+                  streamDefinition: val.streamDefinition,
+                  runtimes: val.runtimes,
+                  logs: val.runtimes.map(runtime => {
+                    let log = '';
+                    if (logs && logs.hasOwnProperty('logs')) {
+                      log = logs.logs[runtime.id] ? logs.logs[runtime.id] : '';
+                    }
+                    return {
+                      id: runtime.id,
+                      log: log
+                    };
+                  })
+                };
+              }));
+          } else {
+            return of({
+              streamDefinition: val.streamDefinition,
+              runtimes: val.runtimes,
+              logs: []
+            });
+          }
+
+        }
+      ))
       .pipe(mergeMap(
         (val: any) => of(Parser.parse(val.streamDefinition.dslText as string, 'stream'))
           .pipe(map((val2) => {
             return {
               streamDefinition: val.streamDefinition,
               runtimes: val.runtimes,
+              logs: val.logs,
               apps: val2.lines[0].nodes
                 .map((node) => ({
                   origin: node['name'],
@@ -95,6 +131,10 @@ export class StreamSummaryComponent implements OnInit {
       );
   }
 
+  changeLog(index) {
+    this.logSelectedIndex = index;
+  }
+
   /**
    * Determine show deployment info
    *

ui/src/app/streams/streams.service.ts

@@ -313,4 +313,18 @@ export class StreamsService {
       );
   }
 
+  /**
+   * Get stream logs
+   * @param {StreamDefinition} stream
+   * @returns {Observable<HttpResponse<any>>}
+   */
+  getLogs(stream: StreamDefinition): Observable<any> {
+    const httpHeaders = HttpUtils.getDefaultHttpHeaders();
+    return this.httpClient
+      .get<any>(`/streams/logs/${stream.name}`, { headers: httpHeaders })
+      .pipe(
+        catchError(this.errorHandler.handleError)
+      );
+  }
+
 }

ui/src/app/tasks/model/task-execution.ts

@@ -13,6 +13,7 @@ export class TaskExecution {
   public jobExecutionIds: number[];
   public errorMessage: string;
   public externalExecutionId: string;
+  public taskExecutionStatus: string;
 
   constructor(executionId: number,
               exitCode: number,
@@ -23,7 +24,8 @@ export class TaskExecution {
               args: string[], // arguments would be restricted name
               jobExecutionIds: number[],
               errorMessage: string,
-              externalExecutionId: string) {
+              externalExecutionId: string,
+              taskExecutionStatus: string) {
     this.executionId = executionId;
     this.exitCode = exitCode;
     this.taskName = taskName;
@@ -33,6 +35,7 @@ export class TaskExecution {
     this.arguments = args;
     this.jobExecutionIds = jobExecutionIds;
     this.errorMessage = errorMessage;
+    this.taskExecutionStatus = taskExecutionStatus;
     this.externalExecutionId = externalExecutionId;
   }
 
@@ -47,7 +50,8 @@ export class TaskExecution {
       jsonItem.arguments,
       jsonItem.jobExecutionIds,
       jsonItem.errorMessage,
-      jsonItem.externalExecutionId);
+      jsonItem.externalExecutionId,
+      jsonItem.taskExecutionStatus);
   }
 
   static pageFromJSON(input): Page<TaskExecution> {

ui/src/app/tasks/task-execution/task-execution.component.html

@@ -1,20 +1,20 @@
-<app-page *ngIf="taskExecution$ | async as taskExecution; else loading">
+<app-page *ngIf="taskExecution$ | async as task; else loading">
 
   <app-page-head>
     <app-page-head-back [defaultUrl]="'/tasks/executions'"
                         [isNotRegex]="'^(\/tasks\/executions\/)'"></app-page-head-back>
     <app-page-head-title>
-      Task Execution Details - <strong>Execution ID: {{ taskExecution.executionId }}</strong>
+      Task Execution Details - <strong>Execution ID: {{ task.task.executionId }}</strong>
     </app-page-head-title>
     <app-page-head-actions>
 
       <button name="task-view" type="button" id="task-view"
-              (click)="detailsTask(taskExecution.taskName)" class="btn btn-primary btn-fa" title="View details task">
+              (click)="detailsTask(task.task.taskName)" class="btn btn-primary btn-fa" title="View details task">
         <span class="fa fa-info-circle"></span>
         View task details
       </button>
       <button name="task-launch" type="button" id="task-launch"
-              (click)="launch(taskExecution.taskName)" class="btn btn-primary btn-fa" title="Relaunch the task"
+              (click)="launch(task.task.taskName)" class="btn btn-primary btn-fa" title="Relaunch the task"
               [dataflowAppRoles]="['ROLE_CREATE']">
         <span class="fa fa-play"></span>
         Relaunch task
@@ -23,7 +23,7 @@
   </app-page-head>
 
   <div dataflowLayoutType type="large">
-    <table *ngIf="taskExecution" class="table table-hover">
+    <table *ngIf="task.task" class="table table-hover">
       <thead>
       <tr>
         <th style="width: 200px">Property</th>
@@ -33,73 +33,81 @@
       <tbody>
       <tr>
         <td>Execution Id</td>
-        <td>{{ taskExecution.executionId }}</td>
+        <td>{{ task.task.executionId }}</td>
       </tr>
       <tr>
         <td>Task Name</td>
         <td>
-          <a [routerLink]="'/tasks/definitions/' + taskExecution.taskName">{{ taskExecution.taskName }}</a>
+          <a [routerLink]="'/tasks/definitions/' + task.task.taskName">{{ task.task.taskName }}</a>
         </td>
       </tr>
       <tr>
         <td>Arguments</td>
         <td>
-          <div *ngIf="taskExecution.arguments">
-            <div *ngFor="let arg of getArguments(taskExecution.arguments)">
+          <div *ngIf="task.task.arguments">
+            <div *ngFor="let arg of getArguments(task.task.arguments)">
               {{ arg[0] }}:
               <strong>{{ arg[1] }}</strong>
             </div>
           </div>
-          <div *ngIf="!taskExecution.arguments">
+          <div *ngIf="!task.task.arguments">
             N/A
           </div>
         </td>
       </tr>
       <tr>
         <td>External Execution Id</td>
-        <td>{{ taskExecution.externalExecutionId || 'N/A' }}</td>
+        <td>{{ task.task.externalExecutionId || 'N/A' }}</td>
       </tr>
       <tr>
         <td>Batch Job</td>
         <td>
-          <span *ngIf="taskExecution.jobExecutionIds.length > 0" class="glyphicon glyphicon-ok text-info"></span>
-          <span *ngIf="taskExecution.jobExecutionIds.length == 0" class="glyphicon glyphicon-remove text-danger"></span>
+          <span *ngIf="task.task.jobExecutionIds.length > 0" class="glyphicon glyphicon-ok text-info"></span>
+          <span *ngIf="task.task.jobExecutionIds.length == 0" class="glyphicon glyphicon-remove text-danger"></span>
         </td>
       </tr>
       <tr>
         <td>Job Execution Ids</td>
-        <td *ngIf="taskExecution.jobExecutionIds.length > 0" id="jobExecutionIds">
+        <td *ngIf="task.task.jobExecutionIds.length > 0" id="jobExecutionIds">
           <a [routerLink]="'/jobs/executions/' + jobExecutionId"
-             *ngFor="let jobExecutionId of taskExecution.jobExecutionIds">
+             *ngFor="let jobExecutionId of task.task.jobExecutionIds">
             {{ jobExecutionId }}
           </a>
         </td>
-        <td *ngIf="taskExecution.jobExecutionIds.length == 0">
+        <td *ngIf="task.task.jobExecutionIds.length == 0">
           N/A
         </td>
       </tr>
       <tr>
         <td>Start Time</td>
-        <td>{{ taskExecution.startTime | dataflowDateTime }}</td>
+        <td>{{ task.task.startTime | dataflowDateTime }}</td>
       </tr>
       <tr>
         <td>End Time</td>
-        <td>{{ taskExecution.endTime | dataflowDateTime }}</td>
+        <td>{{ task.task.endTime | dataflowDateTime }}</td>
       </tr>
       <tr>
         <td>Duration</td>
-        <td>{{ taskExecution.startTime | dataflowDuration: taskExecution.endTime }}</td>
+        <td>{{ task.task.startTime | dataflowDuration: task.task.endTime }}</td>
       </tr>
       <tr>
         <td>Exit Code</td>
-        <td>{{ taskExecution.exitCode }}</td>
+        <td>{{ task.task.exitCode }}</td>
       </tr>
       <tr>
         <td>Exit Message</td>
-        <td>{{ taskExecution.exitMessage || 'N/A' }}</td>
+        <td>{{ task.task.exitMessage || 'N/A' }}</td>
       </tr>
       </tbody>
     </table>
+
+    <div *ngIf="task.logs">
+      <h2 style="padding-bottom: 12px;">Logs</h2>
+      <div class="pui-log">
+        <pre><code>{{ task.logs }}</code></pre>
+      </div>
+    </div>
+
   </div>
 </app-page>
 <ng-template #loading>

ui/src/app/tasks/task-execution/task-execution.component.ts

@@ -2,8 +2,8 @@ import { Component, OnInit } from '@angular/core';
 import { TaskExecution } from '../model/task-execution';
 import { ActivatedRoute, Router } from '@angular/router';
 import { TasksService } from '../tasks.service';
-import { catchError, mergeMap } from 'rxjs/operators';
-import { Observable, EMPTY } from 'rxjs';
+import { catchError, map, mergeMap } from 'rxjs/operators';
+import { Observable, EMPTY, of } from 'rxjs';
 import { RoutingStateService } from '../../shared/services/routing-state.service';
 import { NotificationService } from '../../shared/services/notification.service';
 import { AppError, HttpAppError } from '../../shared/model/error.model';
@@ -24,7 +24,7 @@ export class TaskExecutionComponent implements OnInit {
   /**
    * Observable of Task Execution
    */
-  taskExecution$: Observable<TaskExecution>;
+  taskExecution$: Observable<any>;
 
   /**
    * Constructor
@@ -52,6 +52,25 @@ export class TaskExecutionComponent implements OnInit {
         mergeMap(
           (val) => this.tasksService.getExecution(val.id)
         ),
+        mergeMap(
+          (val) => {
+            if (val.taskExecutionStatus === 'COMPLETE' || val.taskExecutionStatus === 'ERROR') {
+              return this.tasksService.getTaskExecutionLogs(val).pipe(
+                map(logs => {
+                  return {
+                    task: val,
+                    logs: logs
+                  };
+                })
+              );
+            } else {
+              return of({
+                task: val,
+                logs: null
+              });
+            }
+          }
+        ),
         catchError(error => {
           if (HttpAppError.is404(error) || HttpAppError.is400(error)) {
             this.cancel();

ui/src/app/tasks/tasks.service.ts

@@ -16,6 +16,7 @@ import {
 import { HttpResponse } from '@angular/common/http';
 import { Platform, PlatformTask } from '../shared/model/platform';
 import { DataflowEncoder } from '../shared/support/encoder.utils';
+import { StreamDefinition } from '../streams/model/stream-definition';
 
 /**
  * Provides {@link TaskDefinition} related services.
@@ -375,4 +376,18 @@ export class TasksService {
       );
   }
 
+  /**
+   * Get task execution logs
+   * @param {TaskExecution} taskExecution
+   * @returns {Observable<HttpResponse<any>>}
+   */
+  getTaskExecutionLogs(taskExecution: TaskExecution): Observable<any> {
+    const httpHeaders = HttpUtils.getDefaultHttpHeaders();
+    return this.httpClient
+      .get<any>(`/tasks/logs/${taskExecution.externalExecutionId}`, { headers: httpHeaders })
+      .pipe(
+        catchError(this.errorHandler.handleError)
+      );
+  }
+
 }

ui/src/app/tests/mocks/streams.ts

@@ -95,4 +95,8 @@ export class MockStreamsService {
     return of([]);
   }
 
+  getLogs(stream: StreamDefinition): Observable<any> {
+    return of([]);
+  }
+
 }

ui/src/app/tests/mocks/tasks.ts

@@ -122,4 +122,8 @@ export class MockTasksService {
     return of({});
   }
 
+  getTaskExecutionLogs(taskExecution: TaskExecution): Observable<any> {
+    return of('');
+  }
+
 }