Updates and fixes to error reporting

ArroyoSystems · Jun 1, 2023 · 75fea03 · 75fea03
1 parent ff8c75a
commit 75fea03
Show file tree

Hide file tree

Showing 11 changed files with 80 additions and 51 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -11,7 +11,7 @@ members = [
     "arroyo-server-common",
     "arroyo-sql",
     "arroyo-sql-macro",
-    # "arroyo-sql-testing",
+    "arroyo-sql-testing",
     "arroyo-state",
     "arroyo-types",
     "arroyo-worker",

diff --git a/arroyo-api/src/metrics.rs b/arroyo-api/src/metrics.rs
@@ -71,7 +71,9 @@ pub(crate) async fn get_metrics(
                 "{}{{job_id=\"{}\",run_id=\"{}\"}}",
                 TX_QUEUE_REM, job_id, run_id
             );
-            format!("({} - {}) / {}", tx_queue_size, tx_queue_rem, tx_queue_size)
+            // add 1 to each value to account for uninitialized values (which report 0); this can happen when a task
+            // never reads any data
+            format!("1 - (({} + 1) / ({} + 1))", tx_queue_rem, tx_queue_size)
         }
 
         fn get_query(&self, job_id: &str, run_id: u64, rate: &str) -> String {

diff --git a/arroyo-console/src/components/CheckpointDetails.tsx b/arroyo-console/src/components/CheckpointDetails.tsx
@@ -39,7 +39,7 @@ const CheckpointDetails: React.FC<CheckpointDetailsProps> = ({ checkpoint, ops,
   const tableBody = (
     <Tbody>
       {ops?.map(op => {
-        const size = bytes(Object.values(op.checkpoint.tasks));
+        const size = bytes(Object.values(op.checkpoint?.tasks || []));
         const timeline = <OperatorCheckpointTimeline op={op} scale={scale} w={w} />;
         return row(op.name, timeline, size);
       })}

diff --git a/arroyo-console/src/components/Loading.tsx b/arroyo-console/src/components/Loading.tsx
@@ -7,8 +7,8 @@ export interface LoadingProps {
 
 const Loading: React.FC<LoadingProps> = ({ size = 'xl' }) => {
   return (
-    <Flex border={'1px solid blue'} justify={'center'} height={'75%'}>
-      <Spinner speed="0.65s" emptyColor="gray.200" color="blue.500" size={size} />
+    <Flex border={'blue.900'} borderWidth={'1px'} justify={'center'} height={'75%'} p={8}>
+      <Spinner speed="0.65s" emptyColor="gray.200" color="blue.900" size={size} />
     </Flex>
   );
 };

diff --git a/arroyo-console/src/components/OperatorErrors.tsx b/arroyo-console/src/components/OperatorErrors.tsx
@@ -1,18 +1,14 @@
-import { useOperatorErrors } from '../lib/data_fetching';
-import { ApiClient } from '../main';
 import Loading from './Loading';
 import { Table, TableContainer, Tbody, Td, Th, Thead, Tr } from '@chakra-ui/react';
 import React from 'react';
 import { formatDate } from '../lib/util';
+import { OperatorErrorsRes } from '../gen/api_pb';
 
 export interface OperatorErrorsProps {
-  client: ApiClient;
-  jobId?: string;
+  operatorErrors?: OperatorErrorsRes;
 }
 
-const OperatorErrors: React.FC<OperatorErrorsProps> = ({ client, jobId }) => {
-  const { operatorErrors } = useOperatorErrors(client, jobId);
-
+const OperatorErrors: React.FC<OperatorErrorsProps> = ({ operatorErrors }) => {
   if (!operatorErrors) {
     return <Loading />;
   }
@@ -21,7 +17,7 @@ const OperatorErrors: React.FC<OperatorErrorsProps> = ({ client, jobId }) => {
     <Tbody>
       {operatorErrors.messages.map(m => {
         return (
-          <Tr>
+          <Tr key={String(m.createdAt)}>
             <Td>{formatDate(m.createdAt)}</Td>
             <Td>{m.operatorId}</Td>
             <Td>{m.taskIndex?.toString()}</Td>
@@ -34,8 +30,8 @@ const OperatorErrors: React.FC<OperatorErrorsProps> = ({ client, jobId }) => {
   );
 
   const table = (
-    <TableContainer width={'100%'} padding={5}>
-      <Table variant="striped">
+    <TableContainer padding={5}>
+      <Table variant="striped" w={'100%'}>
         <Thead>
           <Tr>
             <Th>Time</Th>

diff --git a/arroyo-console/src/lib/util.ts b/arroyo-console/src/lib/util.ts
@@ -140,7 +140,7 @@ export function getOperatorBackpressure(
 }
 
 export function formatDate(timestamp: bigint) {
-  const date = new Date(Number(timestamp));
+  const date = new Date(Number(timestamp / BigInt(1000)));
   return new Intl.DateTimeFormat('en', {
     dateStyle: 'short',
     timeStyle: 'short',

diff --git a/arroyo-console/src/routes/pipelines/JobDetail.tsx b/arroyo-console/src/routes/pipelines/JobDetail.tsx
@@ -12,6 +12,7 @@ import {
   ButtonGroup,
   Flex,
   Spacer,
+  Spinner,
   Stack,
   Tab,
   TabList,
@@ -31,7 +32,12 @@ import { PipelineOutputs } from './JobOutputs';
 import { CodeEditor } from './SqlEditor';
 import PipelineConfigModal from './PipelineConfigModal';
 import { Code as ConnectWebCode, ConnectError } from '@bufbuild/connect-web';
-import { useJob, useJobCheckpoints, useJobMetrics } from '../../lib/data_fetching';
+import {
+  useJob,
+  useJobCheckpoints,
+  useJobMetrics,
+  useOperatorErrors,
+} from '../../lib/data_fetching';
 import OperatorDetail from '../../components/OperatorDetail';
 import JobNotFound from '../../components/JobNotFound';
 import Checkpoints from '../../components/Checkpoints';
@@ -54,6 +60,7 @@ export function JobDetail({ client }: { client: ApiClient }) {
   const { job, jobError, updateJob } = useJob(client, id);
   const { metrics } = useJobMetrics(client, id);
   const { checkpoints } = useJobCheckpoints(client, id);
+  const { operatorErrors } = useOperatorErrors(client, id);
 
   if (jobError) {
     if (jobError instanceof ConnectError && jobError.code === ConnectWebCode.NotFound) {
@@ -187,33 +194,38 @@ export function JobDetail({ client }: { client: ApiClient }) {
 
     const errorsTab = (
       <TabPanel>
-        <OperatorErrors client={client} jobId={job.jobStatus.jobId} />
+        <OperatorErrors operatorErrors={operatorErrors} />
       </TabPanel>
     );
 
     inner = (
-      <Flex direction={'column'} minH={0}>
-        <Tabs display={'flex'} flexDirection={'column'} overflow={'scroll'}>
-          <div>
-            <TabList>
-              <Tab>Operators</Tab>
-              <Tab>Outputs</Tab>
-              <Tab>Checkpoints</Tab>
-              <Tab>Query</Tab>
-              <Tab>UDFs</Tab>
-              <Tab>Errors</Tab>
-            </TabList>
-          </div>
-          <TabPanels display={'flex'} flexDirection={'column'} flexGrow={1} overflowY={'scroll'}>
-            {operatorsTab}
-            {outputsTab}
-            {checkpointsTab}
-            {queryTab}
-            {udfsTab}
-            {errorsTab}
-          </TabPanels>
-        </Tabs>
-      </Flex>
+      <Tabs h={'100%'}>
+        <div>
+          <TabList>
+            <Tab>Operators</Tab>
+            <Tab>Outputs</Tab>
+            <Tab>Checkpoints</Tab>
+            <Tab>Query</Tab>
+            <Tab>UDFs</Tab>
+            <Tab>
+              Errors{' '}
+              {(operatorErrors?.messages?.length || 0) > 0 && (
+                <Badge ml={2} colorScheme="red" size={'xs'}>
+                  {operatorErrors!.messages.length}
+                </Badge>
+              )}
+            </Tab>
+          </TabList>
+        </div>
+        <TabPanels h={'100%'}>
+          {operatorsTab}
+          {outputsTab}
+          {checkpointsTab}
+          {queryTab}
+          {udfsTab}
+          {errorsTab}
+        </TabPanels>
+      </Tabs>
     );
   }
 
@@ -239,19 +251,18 @@ export function JobDetail({ client }: { client: ApiClient }) {
     actionButton = (
       <Button
         isDisabled={job.action == undefined}
-        isLoading={job.inProgress}
-        loadingText={job.actionText}
         onClick={async () => {
           await updateJobState(job.action!);
         }}
       >
+        {job.inProgress ? <Spinner size="xs" mr={2} /> : null}
         {job.actionText}
       </Button>
     );
   }
 
   return (
-    <Flex direction={'column'} height={'100vh'}>
+    <Box top={0} bottom={0} right={0} left={200} position="absolute" overflowY="hidden">
       <Flex>
         <Box p={5}>
           <Text fontSize={20}>
@@ -268,6 +279,6 @@ export function JobDetail({ client }: { client: ApiClient }) {
       </Flex>
       {inner}
       {configModal}
-    </Flex>
+    </Box>
   );
 }
diff --git a/arroyo-console/src/routes/pipelines/JobGraph.tsx b/arroyo-console/src/routes/pipelines/JobGraph.tsx
@@ -99,8 +99,13 @@ export function PipelineGraph({
   });
 
   return (
-    <Box height={'70vh'} className="pipelineGraph">
-      <ReactFlow nodes={nodes} edges={edges} nodeTypes={nodeTypes}>
+    <Box className="pipelineGraph">
+      <ReactFlow
+        proOptions={{ hideAttribution: true }}
+        nodes={nodes}
+        edges={edges}
+        nodeTypes={nodeTypes}
+      >
         <Background />
       </ReactFlow>
     </Box>

diff --git a/arroyo-sql/src/plan_graph.rs b/arroyo-sql/src/plan_graph.rs
@@ -576,8 +576,6 @@ impl PlanNode {
                         sort_key_type,
                         max_elements: *max_elements,
                     });
-                println!("sliding logical plan operator {:#?}", self);
-                println!("sliding physical plan operator {:#?}", operator);
                 operator
             }
             PlanOperator::TumblingTopN {

diff --git a/arroyo-worker/src/operators/sources/kafka/mod.rs b/arroyo-worker/src/operators/sources/kafka/mod.rs
@@ -124,9 +124,7 @@ where
         let has_state = !state.is_empty();
 
         let state: HashMap<i32, KafkaState> = state.iter().map(|s| (s.partition, **s)).collect();
-        let metadata = consumer
-            .fetch_metadata(Some(&self.topic), Duration::from_secs(30))
-            .expect("failed to fetch kafka metadata");
+        let metadata = consumer.fetch_metadata(Some(&self.topic), Duration::from_secs(30))?;
 
         info!("Fetched metadata for topic {}", self.topic);