[pipeline-ui] Retrieve pod logs from argo archive (#2081)

* Retrieve pod logs from argo archive

* Added aws instance profile iam credential support for minio client. Read workflow status for argo archive location for pod logs.

* fix minor typo, and enforce typing for minio client options

* Update helm chart for pipelines ui role with permission to access secret and workflow crd

* remove unnecessary type cast

* Fix bug: s3client should be a callable, so that iam token is refreshed

frontend/server/aws-helper.ts

@@ -0,0 +1,88 @@
+// Copyright 2019 Google LLC
+//
+// 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
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+import fetch from 'node-fetch';
+
+/** IAWSMetadataCredentials describes the credentials provided by aws metadata store. */
+export interface IAWSMetadataCredentials {
+    Code: string;
+    LastUpdated: string;
+    Type: string;
+    AccessKeyId: string;
+    SecretAccessKey: string;
+    Token: string;
+    Expiration: string;
+}
+
+/** url for aws metadata store. */
+const metadataUrl = "http://169.254.169.254/latest/meta-data/";
+
+
+/**
+ * Get the AWS IAM instance profile.
+ */
+async function getIAMInstanceProfile() : Promise<string|undefined> {
+    try {
+        const resp = await fetch(`${metadataUrl}/iam/security-credentials/`);
+        const profiles = (await resp.text()).split('\n');
+        if (profiles.length > 0) {
+            return profiles[0].trim();  // return first profile
+        }
+        return;
+    } catch (error) {
+        console.error(`Unable to fetch credentials from AWS metadata store: ${error}`)
+        return;
+    }
+}
+
+/**
+ * Class to handle the session credentials for AWS ec2 instance profile.
+ */
+class AWSInstanceProfileCredentials {
+    _iamProfilePromise = getIAMInstanceProfile();
+    _credentials?: IAWSMetadataCredentials;
+    _expiration: number = 0;
+
+    async ok() {
+        return !!(await this._iamProfilePromise);
+    }
+
+    async _fetchCredentials(): Promise<IAWSMetadataCredentials|undefined> {
+        try {
+            const profile = await this._iamProfilePromise;
+            const resp = await fetch(`${metadataUrl}/iam/security-credentials/${profile}`)
+            return resp.json();
+        } catch (error) {
+            console.error(`Unable to fetch credentials from AWS metadata store:${error}`)
+            return;
+        }
+    }
+
+    /**
+     * Get the AWS metadata store session credentials.
+     */
+    async getCredentials(): Promise<IAWSMetadataCredentials> {
+        // query for credentials if going to expire or no credentials yet
+        if ((Date.now() + 10 >= this._expiration) || !this._credentials) {
+            this._credentials = await this._fetchCredentials();
+            if (this._credentials.Expiration)
+                this._expiration = new Date(this._credentials.Expiration).getTime();
+            else 
+                this._expiration = -1;  // always expire
+        }
+        return this._credentials
+    }
+
+}
+
+export const awsInstanceProfileCredentials = new AWSInstanceProfileCredentials();

frontend/server/k8s-helper.ts

@@ -13,10 +13,11 @@
 // limitations under the License.
 
 // @ts-ignore
-import {Core_v1Api, Custom_objectsApi, KubeConfig} from '@kubernetes/client-node';
+import {Core_v1Api, Custom_objectsApi, KubeConfig, V1ConfigMapKeySelector} from '@kubernetes/client-node';
 import * as crypto from 'crypto-js';
 import * as fs from 'fs';
 import * as Utils from './utils';
+import {IPartialArgoWorkflow} from './workflow-helper';
 
 // If this is running inside a k8s Pod, its namespace should be written at this
 // path, this is also how we can tell whether we're running in the cluster.
@@ -30,6 +31,12 @@ const viewerGroup = 'kubeflow.org';
 const viewerVersion = 'v1beta1';
 const viewerPlural = 'viewers';
 
+// Constants for argo workflow
+const workflowGroup = 'argoproj.io'
+const workflowVersion = 'v1alpha1'
+const workflowPlural = 'workflows'
+
+/** Default pod template spec used to create tensorboard viewer. */
 export const defaultPodTemplateSpec = {
   spec: {
     containers: [{
@@ -155,4 +162,38 @@ export function getPodLogs(podName: string): Promise<string> {
       (response: any) => (response && response.body) ? response.body.toString() : '',
       (error: any) => {throw new Error(JSON.stringify(error.body));}
     );
+}
+
+/**
+ * Retrieves the argo workflow CRD.
+ * @param workflowName name of the argo workflow
+ */
+export async function getArgoWorkflow(workflowName: string): Promise<IPartialArgoWorkflow> {
+  if (!k8sV1CustomObjectClient) {
+    throw new Error('Cannot access kubernetes Custom Object API');
+  }
+
+  const res = await k8sV1CustomObjectClient.getNamespacedCustomObject(
+    workflowGroup, workflowVersion, namespace, workflowPlural, workflowName)
+
+  if (res.response.statusCode >= 400) {
+    throw new Error(`Unable to query workflow:${workflowName}: Access denied.`);
+  }
+  return res.body;
+}
+
+/**
+ * Retrieves k8s secret by key and decode from base64.
+ * @param name name of the secret 
+ * @param key key in the secret
+ */
+export async function getK8sSecret(name: string, key: string) {
+  if (!k8sV1Client) {
+    throw new Error('Cannot access kubernetes API');
+  }
+
+  const k8sSecret = await k8sV1Client.readNamespacedSecret(name, namespace);
+  const secretb64 = k8sSecret.body.data[key];
+  const buff = new Buffer(secretb64, 'base64');
+  return buff.toString('ascii');  
 }

frontend/server/minio-helper.ts

@@ -0,0 +1,71 @@
+// Copyright 2019 Google LLC
+//
+// 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
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+import {Stream} from 'stream';
+import * as tar from 'tar';
+import {Client as MinioClient, ClientOptions as MinioClientOptions} from 'minio';
+import {awsInstanceProfileCredentials} from './aws-helper';
+
+
+/** IMinioRequestConfig describes the info required to retrieve an artifact. */
+export interface IMinioRequestConfig {
+    bucket: string;
+    key: string;
+    client: MinioClient;
+}
+
+/** IMinioClientOptionsWithOptionalSecrets wraps around MinioClientOptions where only endPoint is required (accesskey and secretkey are optional). */
+export interface IMinioClientOptionsWithOptionalSecrets extends Partial<MinioClientOptions> {
+  endPoint: string;
+}
+
+/**
+ * Create minio client with aws instance profile credentials if needed.
+ * @param config minio client options where `accessKey` and `secretKey` are optional.
+ */
+export async function createMinioClient(config: IMinioClientOptionsWithOptionalSecrets) {
+
+    if (!config.accessKey || !config.secretKey) {
+        if (await awsInstanceProfileCredentials.ok()) {
+            const credentials = await awsInstanceProfileCredentials.getCredentials();
+            if (credentials) {
+              const {AccessKeyId: accessKey, SecretAccessKey: secretKey, Token: sessionToken} = credentials;
+              return new MinioClient({...config, accessKey, secretKey, sessionToken});
+            }
+            console.error('unable to get credentials from AWS metadata store.')
+        }
+    }
+
+    return new MinioClient(config as MinioClientOptions);
+}
+
+export function getTarObjectAsString({bucket, key, client}: IMinioRequestConfig) {
+    return new Promise<string>(async (resolve, reject) => {
+      try {
+        const stream = await getObjectStream({bucket, key, client});
+        let contents = '';
+        stream.pipe(new tar.Parse()).on('entry', (entry: Stream) => {
+          entry.on('data', (buffer) => contents += buffer.toString());
+        });
+        stream.on('end', () => {
+          resolve(contents);
+        });
+      } catch (err) {
+        reject(err);
+      }
+    });
+}
+
+export function getObjectStream({bucket, key, client}: IMinioRequestConfig) {
+  return client.getObject(bucket, key);
+}

frontend/server/package.json

@@ -11,11 +11,11 @@
     "lodash": ">=4.17.13",
     "minio": "^7.0.0",
     "node-fetch": "^2.1.2",
-    "tar": "^4.4.6"
+    "tar": "^4.4.11"
   },
   "devDependencies": {
     "@types/express": "^4.11.1",
-    "@types/minio": "^6.0.2",
+    "@types/minio": "^7.0.3",
     "@types/node-fetch": "^2.1.2",
     "typescript": "3.3.1"
   },