Add s3 storage artifact route and ui integration of it

chore: Enable fetching markdown from storage for CompareRunsMetricsSection

Add minio dependency

chore: Update minio dependency to version 7.1.3

backend/package.json

@@ -50,6 +50,7 @@
     "http-errors": "^1.8.0",
     "js-yaml": "^4.0.0",
     "lodash": "^4.17.21",
+    "minio": "^7.1.3",
     "pino": "^8.11.0",
     "prom-client": "^14.0.1",
     "ts-node": "^10.9.1"

backend/src/routes/api/storage/index.ts

@@ -0,0 +1,47 @@
+import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
+import { createMinioClient, getObjectStream } from './storageUtils';
+
+export default async (fastify: FastifyInstance): Promise<void> => {
+  fastify.get('/:namespace/:bucket', async (request: FastifyRequest, reply: FastifyReply) => {
+    try {
+      const { namespace, bucket } = request.params as {
+        namespace: string;
+        bucket: string;
+        key: string;
+      };
+      const query = request.query as { [key: string]: string };
+      const key = query.key;
+
+      const stream = await getObjectStream({
+        bucket,
+        client: await createMinioClient(fastify, namespace),
+        key,
+      });
+
+      reply.type('text/plain');
+
+      await new Promise<void>((resolve, reject) => {
+        stream.on('data', (chunk) => {
+          reply.raw.write(chunk);
+        });
+
+        stream.on('end', () => {
+          reply.raw.end();
+          resolve();
+        });
+
+        stream.on('error', (err) => {
+          fastify.log.error('Stream error:', err);
+          reply.raw.statusCode = 500;
+          reply.raw.end(err.message);
+          reject(err);
+        });
+      });
+
+      return;
+    } catch (err) {
+      reply.code(500).send(err.message);
+      return reply;
+    }
+  });
+};

backend/src/routes/api/storage/storageUtils.ts

@@ -0,0 +1,87 @@
+import { Client as MinioClient } from 'minio';
+import { DSPipelineKind, KubeFastifyInstance } from '../../../types';
+import { Transform, PassThrough } from 'stream';
+
+/**
+ * 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(
+  fastify: KubeFastifyInstance,
+  namespace: string,
+): Promise<MinioClient> {
+  try {
+    const dspaResponse = await fastify.kube.customObjectsApi
+      .listNamespacedCustomObject(
+        'datasciencepipelinesapplications.opendatahub.io',
+        'v1alpha1',
+        namespace,
+        'datasciencepipelinesapplications',
+      )
+      .catch((e) => {
+        throw `A ${
+          e.statusCode
+        } error occurred when trying to fetch dspa aws storage credentials: ${
+          e.response?.body?.message || e?.response?.statusMessage
+        }`;
+      });
+
+    const dspas = (
+      dspaResponse?.body as {
+        items: DSPipelineKind[];
+      }
+    )?.items;
+
+    if (!dspas || !dspas.length) {
+      throw 'No Data Science Pipeline Application found';
+    }
+
+    // always get the first one
+    const externalStorage = dspas[0].spec.objectStorage.externalStorage;
+
+    if (externalStorage) {
+      const { region, host: endPoint, s3CredentialsSecret } = externalStorage;
+
+      // get secret
+      const secret = await fastify.kube.coreV1Api.readNamespacedSecret(
+        s3CredentialsSecret.secretName,
+        namespace,
+      );
+      const accessKey = atob(secret.body.data[s3CredentialsSecret.accessKey]);
+      const secretKey = atob(secret.body.data[s3CredentialsSecret.secretKey]);
+
+      if (!accessKey || !secretKey) {
+        throw 'Access key or secret key is empty';
+      }
+
+      // sessionToken
+      return new MinioClient({ accessKey, secretKey, endPoint, region });
+    }
+  } catch (err) {
+    console.error('Unable to create minio client: ', err);
+  }
+}
+
+/** MinioRequestConfig describes the info required to retrieve an artifact. */
+export interface MinioRequestConfig {
+  bucket: string;
+  key: string;
+  client: MinioClient;
+}
+
+/**
+ * Returns a stream from an object in a s3 compatible object store (e.g. minio).
+ *
+ * @param param.bucket Bucket name to retrieve the object from.
+ * @param param.key Key of the object to retrieve.
+ * @param param.client Minio client.
+ *
+ */
+export async function getObjectStream({
+  bucket,
+  key,
+  client,
+}: MinioRequestConfig): Promise<Transform> {
+  const stream = await client.getObject(bucket, key);
+  return stream.pipe(new PassThrough());
+}

backend/src/types.ts

@@ -1014,9 +1014,83 @@ export type K8sCondition = {
   lastHeartbeatTime?: string;
 };
 
+export type DSPipelineExternalStorageKind = {
+  bucket: string;
+  host: string;
+  port?: '';
+  scheme: string;
+  region: string;
+  s3CredentialsSecret: {
+    accessKey: string;
+    secretKey: string;
+    secretName: string;
+  };
+};
+
 export type DSPipelineKind = K8sResourceCommon & {
+  metadata: {
+    name: string;
+    namespace: string;
+  };
   spec: {
     dspVersion: string;
+    apiServer?: Partial<{
+      apiServerImage: string;
+      artifactImage: string;
+      artifactScriptConfigMap: Partial<{
+        key: string;
+        name: string;
+      }>;
+      enableSamplePipeline: boolean;
+    }>;
+    database?: Partial<{
+      externalDB: Partial<{
+        host: string;
+        passwordSecret: Partial<{
+          key: string;
+          name: string;
+        }>;
+        pipelineDBName: string;
+        port: string;
+        username: string;
+      }>;
+      image: string;
+      mariaDB: Partial<{
+        image: string;
+        passwordSecret: Partial<{
+          key: string;
+          name: string;
+        }>;
+        pipelineDBName: string;
+        username: string;
+      }>;
+    }>;
+    mlpipelineUI?: {
+      configMap?: string;
+      image: string;
+    };
+    persistentAgent?: Partial<{
+      image: string;
+      pipelineAPIServerName: string;
+    }>;
+    scheduledWorkflow?: Partial<{
+      image: string;
+    }>;
+    objectStorage: Partial<{
+      externalStorage: DSPipelineExternalStorageKind;
+      minio: Partial<{
+        bucket: string;
+        image: string;
+        s3CredentialsSecret: Partial<{
+          accessKey: string;
+          secretKey: string;
+          secretName: string;
+        }>;
+      }>;
+    }>;
+    viewerCRD?: Partial<{
+      image: string;
+    }>;
   };
   status?: {
     conditions?: K8sCondition[];

frontend/src/concepts/pipelines/content/artifacts/ArtifactUriLink.tsx

@@ -0,0 +1,41 @@
+import React from 'react';
+import { Link } from 'react-router-dom';
+import { usePipelinesAPI } from '~/concepts/pipelines/context';
+import { extractS3UriComponents } from './utils';
+
+interface ArtifactUriLinkProps {
+  uri: string;
+}
+
+export const ArtifactUriLink: React.FC<ArtifactUriLinkProps> = ({ uri }) => {
+  const { namespace } = usePipelinesAPI();
+
+  const url = React.useMemo(() => {
+    // Check if the uri starts with http or https return it as is
+    if (uri.startsWith('http://') || uri.startsWith('https://')) {
+      return uri;
+    }
+
+    // Otherwise check if the uri is s3
+    // If it is not s3, return undefined as we only support fetching from s3 buckets
+    const uriComponents = extractS3UriComponents(uri);
+    if (!uriComponents) {
+      return;
+    }
+
+    const { bucket, path } = uriComponents;
+
+    // /api/storage/${namespace}/${bucket}?key=${path}
+    return `/api/storage/${namespace}/${bucket}?key=${encodeURIComponent(path)}`;
+  }, [namespace, uri]);
+
+  if (!url) {
+    return uri;
+  }
+
+  return (
+    <Link target="_blank" rel="noopener noreferrer" to={url}>
+      {uri}
+    </Link>
+  );
+};

frontend/src/concepts/pipelines/content/artifacts/utils.ts

@@ -0,0 +1,13 @@
+export function extractS3UriComponents(uri: string): { bucket: string; path: string } | undefined {
+  const s3Prefix = 's3://';
+  if (!uri.startsWith(s3Prefix)) {
+    return;
+  }
+
+  const s3UrlWithoutPrefix = uri.slice(s3Prefix.length);
+  const firstSlashIndex = s3UrlWithoutPrefix.indexOf('/');
+  const bucket = s3UrlWithoutPrefix.substring(0, firstSlashIndex);
+  const path = s3UrlWithoutPrefix.substring(firstSlashIndex + 1);
+
+  return { bucket, path };
+}

frontend/src/concepts/pipelines/content/compareRuns/metricsSection/markdown/MarkdownCompare.tsx

@@ -22,6 +22,9 @@ import {
 import { CompareRunsEmptyState } from '~/concepts/pipelines/content/compareRuns/CompareRunsEmptyState';
 import { PipelineRunArtifactSelect } from '~/concepts/pipelines/content/compareRuns/metricsSection/PipelineRunArtifactSelect';
 import MarkdownView from '~/components/MarkdownView';
+import { fetchStorageObject } from '~/services/storageService';
+import { usePipelinesAPI } from '~/concepts/pipelines/context';
+import { extractS3UriComponents } from '~/concepts/pipelines/content/artifacts/utils';
 
 type MarkdownCompareProps = {
   runArtifacts?: RunArtifact[];
@@ -35,6 +38,7 @@ export type MarkdownAndTitle = {
 
 const MarkdownCompare: React.FC<MarkdownCompareProps> = ({ runArtifacts, isLoaded }) => {
   const [expandedGraph, setExpandedGraph] = React.useState<MarkdownAndTitle | undefined>(undefined);
+  const { namespace } = usePipelinesAPI();
 
   const fullArtifactPaths: FullArtifactPath[] = React.useMemo(() => {
     if (!runArtifacts) {
@@ -56,13 +60,25 @@ const MarkdownCompare: React.FC<MarkdownCompareProps> = ({ runArtifacts, isLoade
       }))
       .filter((markdown) => !!markdown.uri)
       .forEach(async ({ uri, title, run }) => {
-        const data = uri; // TODO: fetch data from uri: https://issues.redhat.com/browse/RHOAIENG-7206
+        const uriComponents = extractS3UriComponents(uri);
+        if (!uriComponents) {
+          return;
+        }
+        const text = await fetchStorageObject(
+          namespace,
+          uriComponents.bucket,
+          uriComponents.path,
+        ).catch(() => null);
+
+        if (text === null) {
+          return;
+        }
 
         runMapBuilder[run.run_id] = run;
 
         const config = {
           title,
-          config: data,
+          config: text,
         };
 
         if (run.run_id in configMapBuilder) {
@@ -73,7 +89,7 @@ const MarkdownCompare: React.FC<MarkdownCompareProps> = ({ runArtifacts, isLoade
       });
 
     return { configMap: configMapBuilder, runMap: runMapBuilder };
-  }, [fullArtifactPaths]);
+  }, [fullArtifactPaths, namespace]);
 
   if (!isLoaded) {
     return (

frontend/src/pages/pipelines/global/experiments/artifacts/ArtifactDetails/ArtifactOverviewDetails.tsx

@@ -12,6 +12,7 @@ import {
 } from '@patternfly/react-core';
 
 import { Artifact } from '~/third_party/mlmd';
+import { ArtifactUriLink } from '~/concepts/pipelines/content/artifacts/ArtifactUriLink';
 import { ArtifactPropertyDescriptionList } from './ArtifactPropertyDescriptionList';
 
 interface ArtifactOverviewDetailsProps {
@@ -32,7 +33,9 @@ export const ArtifactOverviewDetails: React.FC<ArtifactOverviewDetailsProps> = (
             {artifact?.uri && (
               <>
                 <DescriptionListTerm>URI</DescriptionListTerm>
-                <DescriptionListDescription>{artifact.uri}</DescriptionListDescription>
+                <DescriptionListDescription>
+                  <ArtifactUriLink uri={artifact.uri} />
+                </DescriptionListDescription>
               </>
             )}
           </DescriptionListGroup>

frontend/src/pages/pipelines/global/experiments/artifacts/ArtifactsTable.tsx

@@ -13,6 +13,7 @@ import SimpleDropdownSelect from '~/components/SimpleDropdownSelect';
 import { ArtifactType } from '~/concepts/pipelines/kfTypes';
 import { useMlmdListContext, usePipelinesAPI } from '~/concepts/pipelines/context';
 import { artifactsDetailsRoute } from '~/routes';
+import { ArtifactUriLink } from '~/concepts/pipelines/content/artifacts/ArtifactUriLink';
 import { FilterOptions, columns, initialFilterData, options } from './constants';
 import { getArtifactName } from './utils';
 
@@ -149,7 +150,9 @@ export const ArtifactsTable: React.FC<ArtifactsTableProps> = ({
         </Td>
         <Td>{artifact.id}</Td>
         <Td>{artifact.type}</Td>
-        <Td>{artifact.uri}</Td>
+        <Td>
+          <ArtifactUriLink uri={artifact.uri} />
+        </Td>
         <Td>
           <PipelinesTableRowTime date={new Date(artifact.createTimeSinceEpoch)} />
         </Td>

frontend/src/pages/pipelines/global/experiments/compareRuns/CompareRunsMetricsSection.tsx

@@ -94,7 +94,6 @@ export const CompareRunMetricsSection: React.FunctionComponent = () => {
         <Tab
           eventKey={MetricSectionTabLabels.MARKDOWN}
           title={<TabTitleText>{MetricSectionTabLabels.MARKDOWN}</TabTitleText>}
-          isDisabled // TODO enable when markdown can be fetched from storage (s3): https://issues.redhat.com/browse/RHOAIENG-7206
         >
           <TabContentBody hasPadding>
             <MarkdownCompare runArtifacts={markdownArtifactData} isLoaded={isLoaded} />