feat: edge sidebar and other fixes (#1110)

Signed-off-by: bbehnke <[email protected]>

ui/src/components/common/SlidingSidebar/partials/EdgeDetails/index.tsx

@@ -1,15 +1,124 @@
-import React from "react";
+import React, { useState, useEffect, useMemo } from "react";
+import Box from "@mui/material/Box";
+import TableContainer from "@mui/material/TableContainer";
+import Table from "@mui/material/Table";
+import TableBody from "@mui/material/TableBody";
+import TableCell from "@mui/material/TableCell";
+import TableHead from "@mui/material/TableHead";
+import TableRow from "@mui/material/TableRow";
+import CircularProgress from "@mui/material/CircularProgress";
+import { usePiplelineWatermarksFetch } from "../../../../../utils/fetchWrappers/piplelineWatermarksFetch";
+import { PipelineWatermarks } from "../../../../../types/declarations/pipeline";
 
 import "./style.css";
 
-
 export interface EdgeDetailsProps {
-  namespaceId: string; // TODO
+  namespaceId: string;
+  pipelineId: string;
   edgeId: string;
 }
 
-export function EdgeDetails({ namespaceId, edgeId }: EdgeDetailsProps) {
+export function EdgeDetails({
+  namespaceId,
+  pipelineId,
+  edgeId,
+}: EdgeDetailsProps) {
+  const [edgeWatermarks, setEdgeWatermarks] = useState<
+    PipelineWatermarks | undefined
+  >();
+  const { data, loading, error } = usePiplelineWatermarksFetch({
+    namespace: namespaceId,
+    pipeline: pipelineId,
+  });
+
+  useEffect(() => {
+    if (!data) {
+      return;
+    }
+    setEdgeWatermarks(data.find((metric) => metric.edgeId === edgeId));
+  }, [data, edgeId]);
+
+  const content = useMemo(() => {
+    if (loading) {
+      return (
+        <Box
+          sx={{ display: "flex", justifyContent: "center", marginTop: "1rem" }}
+        >
+          <CircularProgress />
+        </Box>
+      );
+    }
+    if (error) {
+      return (
+        <Box
+          sx={{ display: "flex", justifyContent: "center", marginTop: "1rem" }}
+        >
+          {`Error loading processing rates: ${error}`}
+        </Box>
+      );
+    }
+    if (!data || !edgeWatermarks) {
+      return (
+        <Box
+          sx={{ display: "flex", justifyContent: "center", marginTop: "1rem" }}
+        >
+          {`No resources found.`}
+        </Box>
+      );
+    }
+    return (
+      <TableContainer
+        sx={{
+          maxHeight: "37.5rem",
+          backgroundColor: "#FFF",
+          marginTop: "1rem",
+        }}
+      >
+        <Table stickyHeader>
+          <TableHead>
+            <TableRow>
+              <TableCell>Partition</TableCell>
+              <TableCell>Watermark</TableCell>
+            </TableRow>
+          </TableHead>
+          <TableBody>
+            {!edgeWatermarks.watermarks.length && (
+              <TableRow>
+                <TableCell colSpan={4} align="center">
+                  No watermarks found
+                </TableCell>
+              </TableRow>
+            )}
+            {!!edgeWatermarks.watermarks.length &&
+              edgeWatermarks.watermarks.map((watermark) => (
+                <TableRow key={watermark.partition}>
+                  <TableCell>{watermark.partition}</TableCell>
+                  <TableCell>{`${watermark.watermark} (${watermark.formattedWatermark})`}</TableCell>
+                </TableRow>
+              ))}
+          </TableBody>
+        </Table>
+      </TableContainer>
+    );
+  }, [data, loading, error, edgeWatermarks]);
+
   return (
-    <div>TODO Edge details</div>
+    <Box
+      sx={{
+        display: "flex",
+        flexDirection: "column",
+        height: "100%",
+      }}
+    >
+      <Box
+        sx={{
+          display: "flex",
+          flexDirection: "row",
+        }}
+      >
+        <span className="edge-details-header-text">{`${edgeId} Edge`}</span>
+      </Box>
+      {content}
+    </Box>
   );
 }

ui/src/components/common/SlidingSidebar/partials/EdgeDetails/style.css

@@ -0,0 +1,5 @@
+.edge-details-header-text {
+  font-size: 1.25rem;
+  font-style: normal;
+  font-weight: 500;
+}

ui/src/components/common/SlidingSidebar/partials/VertexDetails/index.tsx

@@ -6,6 +6,10 @@ import { SpecEditor } from "./partials/SpecEditor";
 import { ProcessingRates } from "./partials/ProcessingRates";
 import { K8sEvents } from "../K8sEvents";
 import { Pods } from "../../../../pages/Pipeline/partials/Graph/partials/NodeInfo/partials/Pods";
+import sourceIcon from "../../../../../images/source_vertex.png";
+import sinkIcon from "../../../../../images/sink_vertex.png";
+import mapIcon from "../../../../../images/map_vertex.png";
+import reducIcon from "../../../../../images/reduce_vertex.png";
 
 import "./style.css";
 
@@ -17,7 +21,8 @@ const K8S_EVENTS_TAB_INDEX = 3;
 export enum VertexType {
   SOURCE,
   SINK,
-  PROCESSOR,
+  MAP,
+  REDUCE,
 }
 
 export interface VertexDetailsProps {
@@ -42,8 +47,10 @@ export function VertexDetails({
     const foundSpec = vertexSpecs.find((spec) => spec.name === vertexId);
     if (foundSpec.source) {
       setVertexType(VertexType.SOURCE);
+    } else if (foundSpec.udf && foundSpec.groupBy) {
+      setVertexType(VertexType.REDUCE);
     } else if (foundSpec.udf) {
-      setVertexType(VertexType.PROCESSOR);
+      setVertexType(VertexType.MAP);
     } else if (foundSpec.sink) {
       setVertexType(VertexType.SINK);
     }
@@ -61,24 +68,51 @@ export function VertexDetails({
     const headerContainerStyle = {
       display: "flex",
       flexDirection: "row",
+      alignItems: "center",
     };
     const textClass = "vertex-details-header-text";
     switch (vertexType) {
       case VertexType.SOURCE:
         return (
           <Box sx={headerContainerStyle}>
+            <img
+              src={sourceIcon}
+              alt="source vertex"
+              className={"vertex-details-header-icon"}
+            />
             <span className={textClass}>Input Vertex</span>
           </Box>
         );
-      case VertexType.PROCESSOR:
+      case VertexType.REDUCE:
         return (
           <Box sx={headerContainerStyle}>
+            <img
+              src={reducIcon}
+              alt="reduce vertex"
+              className={"vertex-details-header-icon"}
+            />
+            <span className={textClass}>Processor Vertex</span>
+          </Box>
+        );
+      case VertexType.MAP:
+        return (
+          <Box sx={headerContainerStyle}>
+            <img
+              src={mapIcon}
+              alt="map vertex"
+              className={"vertex-details-header-icon"}
+            />
             <span className={textClass}>Processor Vertex</span>
           </Box>
         );
       case VertexType.SINK:
         return (
           <Box sx={headerContainerStyle}>
+            <img
+              src={sinkIcon}
+              alt="sink vertex"
+              className={"vertex-details-header-icon"}
+            />
             <span className={textClass}>Sink Vertex</span>
           </Box>
         );

ui/src/components/common/SlidingSidebar/partials/VertexDetails/style.css

@@ -2,6 +2,11 @@
   font-size: 1.25rem;
   font-style: normal;
   font-weight: 500;
+  margin-left: 1rem;
+}
+
+.vertex-details-header-icon {
+  width: 2.25rem;
 }
 
 .vertex-details-tab-panel {

ui/src/components/common/SummaryPageLayout/index.tsx

@@ -41,6 +41,7 @@ export interface SummaryPageLayoutProps {
   collapsedText?: string;
   summarySections: SummarySection[];
   contentComponent: React.ReactNode;
+  contentPadding?: boolean;
 }
 
 const SUMMARY_HEIGHT = "6.5625rem";
@@ -185,6 +186,7 @@ export function SummaryPageLayout({
   collapsedText = "Details",
   summarySections,
   contentComponent,
+  contentPadding = true,
 }: SummaryPageLayoutProps) {
   const [collapsed, setCollapsed] = useState(collapsable && defaultCollapsed);
   const sumaryRef = useRef<any>();
@@ -297,7 +299,7 @@ export function SummaryPageLayout({
       <Box
         sx={{
           marginTop: contentMargin,
-          paddingTop: "1.25rem",
+          paddingTop: contentPadding ? "1.25rem" : "0",
           height: "100%",
         }}
       >

ui/src/components/pages/Pipeline/index.tsx

@@ -133,6 +133,8 @@ export function Pipeline() {
 
   return (
     <SummaryPageLayout
+      contentPadding={false}
+      collapsable
       summarySections={summarySections}
       contentComponent={
         <div

ui/src/components/pages/Pipeline/partials/Graph/index.tsx

@@ -351,12 +351,13 @@ export default function Graph(props: GraphProps) {
           type: SidebarType.EDGE_DETAILS,
           edgeDetailsProps: {
             namespaceId,
+            pipelineId,
             edgeId: edge.id,
           },
         });
       }
     },
-    [setSidebarProps, namespaceId]
+    [setSidebarProps, namespaceId, pipelineId]
   );
 
   // This has been added to make sure that edge container refreshes on edges being refreshed

ui/src/types/declarations/pipeline.d.ts

@@ -129,3 +129,26 @@ export interface PiplelineVertexMetricsFetchProps {
   pipeline: string;
   loadOnRefresh?: boolean;
 }
+
+export interface PipelineWatermark {
+  partition: number;
+  watermark: number;
+  formattedWatermark: string;
+}
+
+export interface PipelineWatermarks {
+  edgeId: string;
+  watermarks: PipelineWatermark[];
+}
+
+export interface PipelineWatermarksFetchResult {
+  data?: PipelineWatermarks[];
+  loading: boolean;
+  error: any;
+}
+
+export interface PipelineWatermarksFetchProps {
+  namespace: string;
+  pipeline: string;
+  loadOnRefresh?: boolean;
+}

ui/src/utils/fetchWrappers/piplelineVertexMetricsFetch.ts

@@ -18,12 +18,21 @@ const rawDataToVertexMetrics = (
       metrics: rawData[vertexId].map((item: any, index: number) => {
         return {
           partition: index,
-          oneM: item.processingRates["1m"] || 0,
-          fiveM: item.processingRates["5m"] || 0,
-          fifteenM: item.processingRates["15m"] || 0,
+          oneM:
+            item.processingRates && item.processingRates["1m"]
+              ?item.processingRates["1m"].toFixed(2)
+              : 0,
+          fiveM:
+            item.processingRates && item.processingRates["5m"]
+              ? item.processingRates["5m"].toFixed(2)
+              : 0,
+          fifteenM:
+            item.processingRates && item.processingRates["15m"]
+              ? item.processingRates["15m"].toFixed(2)
+              : 0,
         };
       }),
-    }
+    };
   });
 };
 
@@ -84,13 +93,7 @@ export const usePiplelineVertexMetricsFetch = ({
       });
       return;
     }
-  }, [
-    data,
-    loading,
-    error,
-    loadOnRefresh,
-    options,
-  ]);
+  }, [data, loading, error, loadOnRefresh, options]);
 
   return results;
 };