Add pipeline edit modal with parallelism control

arroyo-console/src/routes/pipelines/JobDetail.tsx

@@ -11,6 +11,7 @@ import {
   Badge,
   Box,
   Button,
+  ButtonGroup,
   chakra,
   Code,
   Flex,
@@ -41,6 +42,7 @@ import {
   theme,
   Tr,
   UnorderedList,
+  useDisclosure
 } from "@chakra-ui/react";
 import React, { ReactElement, ReactNode, useCallback, useEffect, useMemo, useRef, useState } from "react";
 import {
@@ -76,6 +78,7 @@ import { PipelineGraph } from "./JobGraph";
 import { ApiClient } from "../../main";
 import { PipelineOutputs } from "./JobOutputs";
 import { SqlEditor } from "./SqlEditor";
+import PipelineConfigModal from "./PipelineConfigModal";
 
 interface JobDetailState {
   pipeline?: JobDetailsResp;
@@ -95,6 +98,11 @@ export function JobDetail({ client }: { client: ApiClient }) {
   const [checkpoints, setCheckpoints] = useState<Array<CheckpointOverview>>([]);
   const [checkpointFetchInProgress, setCheckpointFetchInProgress] = useState<boolean>(false);
   const [subscribed, setSubscribed] = useState<boolean>(false);
+  const {
+    isOpen: configModalOpen,
+    onOpen: onConfigModalOpen,
+    onClose: onConfigModalClose
+  } = useDisclosure();
 
   let { id } = useParams();
 
@@ -186,6 +194,15 @@ export function JobDetail({ client }: { client: ApiClient }) {
     fetchPipeline();
   }
 
+  async function updateJobParallelism (parallelism: number) {
+    console.log(`Setting pipeline parallelism=${parallelism}`);
+    await (await client()).updateJob({
+      jobId: id,
+      parallelism
+    });
+    fetchPipeline();
+  }
+
   let inner;
   if (state.pipeline == null || state.pipeline.jobGraph == null) {
     inner = <p>Loading</p>;
@@ -274,7 +291,35 @@ export function JobDetail({ client }: { client: ApiClient }) {
     );
   }
 
+  let configModal = <></>;
+  if (state.pipeline?.jobGraph?.nodes) {
+    const { nodes } = state.pipeline.jobGraph;
+    const parallelism = Math.max(...nodes.map(({ parallelism }) => parallelism));
+
+    configModal = (
+      <PipelineConfigModal
+        isOpen={configModalOpen}
+        parallelism={parallelism}
+        onClose={onConfigModalClose}
+        updateJobParallelism={updateJobParallelism}
+      />
+    );
+  }
+
+  const editPipelineButton = <Button onClick={onConfigModalOpen}>Edit</Button>;
 
+  const actionButton = (
+    <Button
+      isDisabled={state.pipeline?.action == undefined}
+      isLoading={state.pipeline?.inProgress}
+      loadingText={state.pipeline?.actionText}
+      onClick={async () => {
+        await updateJobState(state.pipeline?.action!);
+      }}
+    >
+      {state.pipeline?.actionText}
+    </Button>
+  );
 
   return (
     <Box top={0} bottom={0} right={0} left={200} position="absolute" overflowY="hidden">
@@ -286,17 +331,14 @@ export function JobDetail({ client }: { client: ApiClient }) {
         </Box>
         <Spacer />
         <Box p={5}>
-          <Button
-            isDisabled={state.pipeline?.action == undefined}
-            isLoading={state.pipeline?.inProgress}
-            loadingText={state.pipeline?.actionText}
-            onClick={() => updateJobState(state.pipeline?.action!)}
-          >
-            {state.pipeline?.actionText}
-          </Button>
+          <ButtonGroup>
+            {editPipelineButton}
+            {actionButton}
+          </ButtonGroup>
         </Box>
       </Flex>
       {inner}
+      {configModal}
     </Box>
   );
 }

arroyo-console/src/routes/pipelines/PipelineConfigModal.tsx

@@ -0,0 +1,133 @@
+import React, { useState } from "react";
+import {
+  Alert,
+  AlertDescription,
+  AlertIcon,
+  Button,
+  FormControl,
+  FormErrorMessage,
+  FormHelperText,
+  FormLabel,
+  Modal,
+  ModalBody,
+  ModalCloseButton,
+  ModalContent,
+  ModalFooter,
+  ModalHeader,
+  ModalOverlay,
+  NumberDecrementStepper,
+  NumberIncrementStepper,
+  NumberInput,
+  NumberInputField,
+  NumberInputStepper,
+} from "@chakra-ui/react";
+
+export interface PipelineConfigModalProps {
+  isOpen: boolean;
+  parallelism: number;
+  onClose: () => void;
+  updateJobParallelism: (parallelism: number) => void;
+}
+
+const MAX_PARALLELISM = 10;
+
+const PipelineConfigModal: React.FC<PipelineConfigModalProps> = ({
+  isOpen,
+  parallelism,
+  onClose,
+  updateJobParallelism,
+}) => {
+  const [parallelismInputValue, setParallelismInputValue] = useState<number | undefined>(
+    parallelism
+  );
+
+  const close = () => {
+    onClose();
+
+    // reset state
+    setParallelismInputValue(parallelism);
+  };
+
+  const onConfirm = () => {
+    if (parallelismInputValue) {
+      updateJobParallelism(parallelismInputValue);
+    }
+    onClose();
+  };
+
+  const pipelineRequiresRestart = parallelismInputValue != parallelism;
+
+  let changeAlert = <></>;
+  if (pipelineRequiresRestart) {
+    changeAlert = (
+      <Alert status="warning">
+        <AlertIcon />
+        <AlertDescription>These changes require a pipeline restart.</AlertDescription>
+      </Alert>
+    );
+  }
+
+  const parallelismInput = (
+    <NumberInput
+      isRequired
+      min={1}
+      max={MAX_PARALLELISM}
+      value={parallelismInputValue ?? ""}
+      onChange={(valueAsString, valueAsNumber) => {
+        if (isNaN(valueAsNumber)) {
+          setParallelismInputValue(undefined);
+        } else {
+          setParallelismInputValue(valueAsNumber);
+        }
+      }}
+    >
+      <NumberInputField bg="gray.800" />
+      <NumberInputStepper>
+        <NumberIncrementStepper />
+        <NumberDecrementStepper />
+      </NumberInputStepper>
+    </NumberInput>
+  );
+
+  let parallelismHasError = false;
+  let parallelismErrorText = "";
+  let saveEnabled: boolean;
+
+  if (!parallelismInputValue) {
+    parallelismHasError = true;
+    parallelismErrorText = "Enter a valid number.";
+  }
+
+  saveEnabled = !parallelismHasError;
+
+  const parallelismForm = (
+    <FormControl isRequired isInvalid={parallelismHasError}>
+      <FormLabel>Parallelism</FormLabel>
+      {parallelismInput}
+      <FormHelperText>Number of parallel substacks for each node (maximum 10)</FormHelperText>
+      <FormErrorMessage>{parallelismErrorText}</FormErrorMessage>
+    </FormControl>
+  );
+
+  return (
+    <Modal isOpen={isOpen} onClose={close}>
+      <ModalOverlay />
+      <ModalContent>
+        <ModalHeader>Edit Pipeline Configuration</ModalHeader>
+        <ModalCloseButton />
+        {changeAlert}
+        <ModalBody>{parallelismForm}</ModalBody>
+        <ModalFooter>
+          <Button variant="ghost" onClick={onClose}>
+            Cancel
+          </Button>
+          <Button colorScheme="blue" mr={3} isDisabled={!saveEnabled} onClick={onConfirm}>
+            Save
+          </Button>
+        </ModalFooter>
+      </ModalContent>
+    </Modal>
+  );
+};
+
+export default PipelineConfigModal;