feat(frontend): support tensorboard viewer and other visualize Result…

…s using volume mount path. Part of kubeflow#4208 (kubeflow#4236)

* support local file storage type for local volume mount path, refer: kubeflow#4208

* add todo comment to support directory and filePath include wildcards '*', detail refer: kubeflow#4208

* revert old code indent

* run 'npm run format' to format code

* support tensorboard viewer and other visualize Results using volume mount path, modify 'file' schema to 'volume':

1. source schema: volume://volume-name/relative/path/from/volume/xxx.csv
2. for tensorboard(also support Series1:volume://volume-name/path_to_model_dir_1,Series2:volume://volume-name/path_to_model_dir_2):
* check volume-name was specified in podTemplateSpec( which was inject by VIEWER_TENSORBOARD_POD_TEMPLATE_SPEC_PATH env)
* check /relative/path/from/volume/xxx file path was prefix-mounted in podTemplateSpec
3. for others:
* check volume-name was specified at ml-pipeline-ui pod
* check /relative/path/from/volume/xxx.csv file path exist

* fix test and add more tests

* change error message not found to not exist.

* fix tensorboard create test

* combining volume mount path and key as artifacts path

* extra complex code to a function and add more test

* use ml-pipeline-ui container name to find server container instead of use containers[0]

* fix review suggestion: kubeflow#4236

* format code

* extract how to find file path on a pod volume to a common function, and optimize error message

* fix k8s-helper.test error

* add more documentation and fix mistake: volumeMountPath to filePathInVolume

* fix test error

* Update k8s-helper.test.ts

* format error message

Co-authored-by: Yuan (Bob) Gong <gongyuan94@gmail.com>

frontend/server/handlers/artifacts.ts

@@ -14,20 +14,24 @@
 import fetch from 'node-fetch';
 import { AWSConfigs, HttpConfigs, MinioConfigs, ProcessEnv } from '../configs';
 import { Client as MinioClient } from 'minio';
-import { PreviewStream } from '../utils';
+import { PreviewStream, findFileOnPodVolume } from '../utils';
 import { createMinioClient, getObjectStream } from '../minio-helper';
+import * as serverInfo from '../helpers/server-info';
 import { Handler, Request, Response } from 'express';
 import { Storage } from '@google-cloud/storage';
 import proxy from 'http-proxy-middleware';
 import { HACK_FIX_HPM_PARTIAL_RESPONSE_HEADERS } from '../consts';
 
+import * as fs from 'fs';
+import { V1Container } from '@kubernetes/client-node/dist/api';
+
 /**
  * ArtifactsQueryStrings describes the expected query strings key value pairs
  * in the artifact request object.
  */
 interface ArtifactsQueryStrings {
   /** artifact source. */
-  source: 'minio' | 's3' | 'gcs' | 'http' | 'https';
+  source: 'minio' | 's3' | 'gcs' | 'http' | 'https' | 'volume';
   /** bucket name. */
   bucket: string;
   /** artifact key/path that is uri encoded.  */
@@ -101,6 +105,16 @@ export function getArtifactsHandler(artifactsConfigs: {
         )(req, res);
         break;
 
+      case 'volume':
+        await getVolumeArtifactsHandler(
+          {
+            bucket,
+            key,
+          },
+          peek,
+        )(req, res);
+        break;
+
       default:
         res.status(500).send('Unknown storage source: ' + source);
         return;
@@ -240,6 +254,53 @@ function getGCSArtifactHandler(options: { key: string; bucket: string }, peek: n
   };
 }
 
+function getVolumeArtifactsHandler(options: { bucket: string; key: string }, peek: number = 0) {
+  const { key, bucket } = options;
+  return async (req: Request, res: Response) => {
+    try {
+      const [pod, err] = await serverInfo.getHostPod();
+      if (err) {
+        res.status(500).send(err);
+        return;
+      }
+
+      if (!pod) {
+        res.status(500).send('Could not get server pod');
+        return;
+      }
+
+      // ml-pipeline-ui server container name also be called 'ml-pipeline-ui-artifact' in KFP multi user mode.
+      // https://github.com/kubeflow/manifests/blob/master/pipeline/installs/multi-user/pipelines-profile-controller/sync.py#L212
+      const [filePath, parseError] = findFileOnPodVolume(pod, {
+        containerNames: ['ml-pipeline-ui', 'ml-pipeline-ui-artifact'],
+        volumeMountName: bucket,
+        filePathInVolume: key,
+      });
+      if (parseError) {
+        res.status(404).send(`Failed to open volume://${bucket}/${key}, ${parseError}`);
+        return;
+      }
+
+      // TODO: support directory and support filePath include wildcards '*'
+      const stat = await fs.promises.stat(filePath);
+      if (stat.isDirectory()) {
+        res
+          .status(400)
+          .send(
+            `Failed to open volume://${bucket}/${key}, file ${filePath} is directory, does not support now`,
+          );
+        return;
+      }
+
+      fs.createReadStream(filePath)
+        .pipe(new PreviewStream({ peek }))
+        .pipe(res);
+    } catch (err) {
+      res.status(500).send(`Failed to open volume://${bucket}/${key}: ${err}`);
+    }
+  };
+}
+
 const ARTIFACTS_PROXY_DEFAULTS = {
   serviceName: 'ml-pipeline-ui-artifact',
   servicePort: '80',

frontend/server/helpers/server-info.ts

@@ -0,0 +1,55 @@
+// Copyright 2019-2020 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 * as fs from 'fs';
+import { V1Pod } from '@kubernetes/client-node';
+import { getPod } from '../k8s-helper';
+
+const namespaceFilePath = '/var/run/secrets/kubernetes.io/serviceaccount/namespace';
+let serverNamespace: string | undefined;
+let hostPod: V1Pod | undefined;
+
+// The file path contains pod namespace when in Kubernetes cluster.
+if (fs.existsSync(namespaceFilePath)) {
+  serverNamespace = fs.readFileSync(namespaceFilePath, 'utf-8');
+}
+
+// get ml-pipeline-ui host pod
+export async function getHostPod(): Promise<[V1Pod | undefined, undefined] | [undefined, string]> {
+  // use cached hostPod
+  if (hostPod) {
+    return [hostPod, undefined];
+  }
+
+  if (!serverNamespace) {
+    return [undefined, "server namespace can't be obtained"];
+  }
+
+  // get ml-pipeline-ui server pod name
+  const { HOSTNAME: POD_NAME } = process.env;
+  if (!POD_NAME) {
+    return [undefined, "server pod name can't be obtained"];
+  }
+
+  const [pod, err] = await getPod(POD_NAME, serverNamespace);
+
+  if (err) {
+    const { message, additionalInfo } = err;
+    console.error(message, additionalInfo);
+    return [undefined, `Failed to get host pod: ${message}`];
+  }
+
+  hostPod = pod;
+  return [hostPod, undefined];
+}

frontend/server/k8s-helper.test.ts

@@ -0,0 +1,167 @@
+// Copyright 2020 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 { TEST_ONLY as K8S_TEST_EXPORT } from './k8s-helper';
+
+describe('k8s-helper', () => {
+  describe('parseTensorboardLogDir', () => {
+    const podTemplateSpec = {
+      spec: {
+        containers: [
+          {
+            volumeMounts: [
+              {
+                name: 'output',
+                mountPath: '/data',
+              },
+              {
+                name: 'artifact',
+                subPath: 'pipeline1',
+                mountPath: '/data1',
+              },
+              {
+                name: 'artifact',
+                subPath: 'pipeline2',
+                mountPath: '/data2',
+              },
+            ],
+          },
+        ],
+        volumes: [
+          {
+            name: 'output',
+            hostPath: {
+              path: '/data/output',
+              type: 'Directory',
+            },
+          },
+          {
+            name: 'artifact',
+            persistentVolumeClaim: {
+              claimName: 'artifact_pvc',
+            },
+          },
+        ],
+      },
+    };
+
+    it('handles not volume storage', () => {
+      const logdir = 'gs://testbucket/test/key/path';
+      const url = K8S_TEST_EXPORT.parseTensorboardLogDir(logdir, podTemplateSpec);
+      expect(url).toEqual(logdir);
+    });
+
+    it('handles not volume storage with Series', () => {
+      const logdir =
+        'Series1:gs://testbucket/test/key/path1,Series2:gs://testbucket/test/key/path2';
+      const url = K8S_TEST_EXPORT.parseTensorboardLogDir(logdir, podTemplateSpec);
+      expect(url).toEqual(logdir);
+    });
+
+    it('handles volume storage without subPath', () => {
+      const logdir = 'volume://output';
+      const url = K8S_TEST_EXPORT.parseTensorboardLogDir(logdir, podTemplateSpec);
+      expect(url).toEqual('/data');
+    });
+
+    it('handles volume storage without subPath with Series', () => {
+      const logdir = 'Series1:volume://output/volume/path1,Series2:volume://output/volume/path2';
+      const url = K8S_TEST_EXPORT.parseTensorboardLogDir(logdir, podTemplateSpec);
+      expect(url).toEqual('Series1:/data/volume/path1,Series2:/data/volume/path2');
+    });
+
+    it('handles volume storage with subPath', () => {
+      const logdir = 'volume://artifact/pipeline1';
+      const url = K8S_TEST_EXPORT.parseTensorboardLogDir(logdir, podTemplateSpec);
+      expect(url).toEqual('/data1');
+    });
+
+    it('handles volume storage with subPath with Series', () => {
+      const logdir =
+        'Series1:volume://artifact/pipeline1/path1,Series2:volume://artifact/pipeline2/path2';
+      const url = K8S_TEST_EXPORT.parseTensorboardLogDir(logdir, podTemplateSpec);
+      expect(url).toEqual('Series1:/data1/path1,Series2:/data2/path2');
+    });
+
+    it('handles volume storage without subPath throw volume not configured error', () => {
+      const logdir = 'volume://other/path';
+      expect(() => K8S_TEST_EXPORT.parseTensorboardLogDir(logdir, podTemplateSpec)).toThrowError(
+        'Cannot find file "volume://other/path" in pod "unknown": volume "other" not configured',
+      );
+    });
+
+    it('handles volume storage without subPath throw volume not configured error with Series', () => {
+      const logdir = 'Series1:volume://output/volume/path1,Series2:volume://other/volume/path2';
+      expect(() => K8S_TEST_EXPORT.parseTensorboardLogDir(logdir, podTemplateSpec)).toThrowError(
+        'Cannot find file "volume://other/volume/path2" in pod "unknown": volume "other" not configured',
+      );
+    });
+
+    it('handles volume storage without subPath throw volume not mounted', () => {
+      const noMountPodTemplateSpec = {
+        spec: {
+          volumes: [
+            {
+              name: 'artifact',
+              persistentVolumeClaim: {
+                claimName: 'artifact_pvc',
+              },
+            },
+          ],
+        },
+      };
+      const logdir = 'volume://artifact/path1';
+      expect(() =>
+        K8S_TEST_EXPORT.parseTensorboardLogDir(logdir, noMountPodTemplateSpec),
+      ).toThrowError(
+        'Cannot find file "volume://artifact/path1" in pod "unknown": container "" not found',
+      );
+    });
+
+    it('handles volume storage without volumeMounts throw volume not mounted', () => {
+      const noMountPodTemplateSpec = {
+        spec: {
+          containers: [
+            {
+              volumeMounts: [
+                {
+                  name: 'other',
+                  mountPath: '/data',
+                },
+              ],
+            },
+          ],
+          volumes: [
+            {
+              name: 'artifact',
+              persistentVolumeClaim: {
+                claimName: 'artifact_pvc',
+              },
+            },
+          ],
+        },
+      };
+      const logdir = 'volume://artifact/path';
+      expect(() => K8S_TEST_EXPORT.parseTensorboardLogDir(logdir, podTemplateSpec)).toThrowError(
+        'Cannot find file "volume://artifact/path" in pod "unknown": volume "artifact" not mounted',
+      );
+    });
+
+    it('handles volume storage with subPath throw volume mount not found', () => {
+      const logdir = 'volume://artifact/other';
+      expect(() => K8S_TEST_EXPORT.parseTensorboardLogDir(logdir, podTemplateSpec)).toThrowError(
+        'Cannot find file "volume://artifact/other" in pod "unknown": volume "artifact" not mounted or volume "artifact" with subPath (which is prefix of other) not mounted',
+      );
+    });
+  });
+});