844 toggle on and off of the extractor livelihood (#869)

longshuicy · web-flow · commit 6cf0037dc5c3 · 2024-01-18T14:16:02.000-06:00
* working

* toggle logic works

* styling of the switch component

* set back

* black format

* update aggregated filters

* add more comments and docstrings

* black format
diff --git a/backend/app/routers/listeners.py b/backend/app/routers/listeners.py
@@ -91,6 +91,31 @@ async def _check_livelihood(
         return True
 
 
+def _check_livelihood_query(heartbeat_interval=settings.listener_heartbeat_interval):
+    """
+    Return a MongoDB aggregation query to check the livelihood of a listener.
+    This is needed due to the alive field being a computed field.
+    When a user call the /listeners endpoint, the alive field will be computed with the current datetime
+    and compared with heartbeat_interval.
+    """
+    if heartbeat_interval == 0:
+        heartbeat_interval = settings.listener_heartbeat_interval
+
+    # Perform aggregation queries that adds alive flag using the PyMongo syntax
+    aggregated_query = {
+        "$addFields": {
+            "alive": {
+                "$lt": [
+                    {"$subtract": [datetime.datetime.utcnow(), "$lastAlive"]},
+                    heartbeat_interval * 1000,  # convert to milliseconds
+                ]
+            }
+        }
+    }
+
+    return aggregated_query
+
+
 @router.get("/instance")
 async def get_instance_id(
     user=Depends(get_current_user),
@@ -246,33 +271,40 @@ async def get_listeners(
     heartbeat_interval: Optional[int] = settings.listener_heartbeat_interval,
     category: Optional[str] = None,
     label: Optional[str] = None,
+    alive_only: Optional[bool] = False,
 ):
     """Get a list of all Event Listeners in the db.
 
     Arguments:
         skip -- number of initial records to skip (i.e. for pagination)
         limit -- restrict number of records to be returned (i.e. for pagination)
+        heartbeat_interval -- number of seconds after which a listener is considered dead
         category -- filter by category has to be exact match
         label -- filter by label has to be exact match
+        alive_only -- filter by alive status
     """
-    query = []
+    # First compute alive flag for all listeners
+    aggregation_pipeline = [
+        _check_livelihood_query(heartbeat_interval=heartbeat_interval)
+    ]
+
+    # Add filters if applicable
     if category:
-        query.append(EventListenerDB.properties.categories == category)
+        aggregation_pipeline.append({"$match": {"properties.categories": category}})
     if label:
-        query.append(EventListenerDB.properties.default_labels == label)
+        aggregation_pipeline.append({"$match": {"properties.default_labels": label}})
+    if alive_only:
+        aggregation_pipeline.append({"$match": {"alive": True}}),
 
-    # sort by name alphabetically
-    listeners = await EventListenerDB.find(
-        *query, skip=skip, limit=limit, sort=EventListenerDB.name
-    ).to_list()
+    # Sort by name alphabetically and then pagination
+    aggregation_pipeline.append({"$sort": {"name": 1}})
+    aggregation_pipeline.append({"$skip": skip})
+    aggregation_pipeline.append({"$limit": limit})
 
-    # batch return listener statuses for easy consumption
-    listener_response = []
-    for listener in listeners:
-        listener.alive = await _check_livelihood(listener, heartbeat_interval)
-        listener_response.append(listener.dict())
+    # Run aggregate query and return
+    listeners = await EventListenerDB.aggregate(aggregation_pipeline).to_list()
 
-    return listener_response
+    return listeners
 
 
 @router.put("/{listener_id}", response_model=EventListenerOut)
diff --git a/frontend/src/actions/listeners.js b/frontend/src/actions/listeners.js
@@ -8,7 +8,8 @@ export function fetchListeners(
 	limit = 21,
 	heartbeatInterval = 0,
 	category = null,
-	label = null
+	label = null,
+	aliveOnly = false
 ) {
 	return (dispatch) => {
 		// TODO: Parameters for dates? paging?
@@ -17,7 +18,8 @@ export function fetchListeners(
 			limit,
 			heartbeatInterval,
 			category,
-			label
+			label,
+			aliveOnly
 		)
 			.then((json) => {
 				dispatch({
@@ -30,7 +32,14 @@ export function fetchListeners(
 				dispatch(
 					handleErrors(
 						reason,
-						fetchListeners(skip, limit, heartbeatInterval, category, label)
+						fetchListeners(
+							skip,
+							limit,
+							heartbeatInterval,
+							category,
+							label,
+							aliveOnly
+						)
 					)
 				);
 			});
diff --git a/frontend/src/components/listeners/ExtractionHistory.tsx b/frontend/src/components/listeners/ExtractionHistory.tsx
@@ -80,15 +80,17 @@ export const ExtractionHistory = (): JSX.Element => {
 		limit: number | undefined,
 		heartbeatInterval: number | undefined,
 		selectedCategory: string | null,
-		selectedLabel: string | null
+		selectedLabel: string | null,
+		aliveOnly: boolean | undefined
 	) =>
 		dispatch(
 			fetchListeners(
 				skip,
 				limit,
 				heartbeatInterval,
 				selectedCategory,
-				selectedLabel
+				selectedLabel,
+				aliveOnly
 			)
 		);
 	const listListenerJobs = (
@@ -128,9 +130,10 @@ export const ExtractionHistory = (): JSX.Element => {
 	const [executionJobsTableRow, setExecutionJobsTableRow] = useState([]);
 	const [selectedStatus, setSelectedStatus] = useState(null);
 	const [selectedCreatedTime, setSelectedCreatedTime] = useState(null);
+	const [aliveOnly, setAliveOnly] = useState<boolean>(false);
 
 	useEffect(() => {
-		listListeners(skip, limit, 0, null, null);
+		listListeners(skip, limit, 0, null, null, aliveOnly);
 		listListenerJobs(null, null, null, null, null, null, 0, 100);
 	}, []);
 
@@ -197,7 +200,7 @@ export const ExtractionHistory = (): JSX.Element => {
 
 	useEffect(() => {
 		if (skip !== null && skip !== undefined) {
-			listListeners(skip, limit, 0, null, null);
+			listListeners(skip, limit, 0, null, null, aliveOnly);
 			if (skip === 0) setPrevDisabled(true);
 			else setPrevDisabled(false);
 		}
diff --git a/frontend/src/components/listeners/Listeners.tsx b/frontend/src/components/listeners/Listeners.tsx
@@ -5,12 +5,14 @@ import {
 	ButtonGroup,
 	Divider,
 	FormControl,
+	FormControlLabel,
 	Grid,
 	InputLabel,
 	List,
 	MenuItem,
 	Paper,
 	Select,
+	Switch,
 } from "@mui/material";
 
 import { RootState } from "../../types/data";
@@ -42,15 +44,17 @@ export function Listeners(props: ListenerProps) {
 		limit: number | undefined,
 		heartbeatInterval: number | undefined,
 		selectedCategory: string | null,
-		selectedLabel: string | null
+		selectedLabel: string | null,
+		aliveOnly: boolean | undefined
 	) =>
 		dispatch(
 			fetchListeners(
 				skip,
 				limit,
 				heartbeatInterval,
 				selectedCategory,
-				selectedLabel
+				selectedLabel,
+				aliveOnly
 			)
 		);
 	const searchListeners = (
@@ -81,10 +85,11 @@ export function Listeners(props: ListenerProps) {
 	const [searchText, setSearchText] = useState<string>("");
 	const [selectedCategory, setSelectedCategory] = useState("");
 	const [selectedLabel, setSelectedLabel] = useState("");
+	const [aliveOnly, setAliveOnly] = useState<boolean>(false);
 
 	// component did mount
 	useEffect(() => {
-		listListeners(skip, limit, 0, null, null);
+		listListeners(skip, limit, 0, null, null, aliveOnly);
 		listAvailableCategories();
 		listAvailableLabels();
 	}, []);
@@ -101,13 +106,20 @@ export function Listeners(props: ListenerProps) {
 		if (searchText !== "") {
 			handleListenerSearch();
 		} else {
-			listListeners(skip, limit, 0, selectedCategory, selectedLabel);
+			listListeners(skip, limit, 0, selectedCategory, selectedLabel, aliveOnly);
 		}
 	}, [searchText]);
 
+	useEffect(() => {
+		setSearchText("");
+		// flip to first page
+		setSkip(0);
+		listListeners(0, limit, 0, selectedCategory, selectedLabel, aliveOnly);
+	}, [aliveOnly]);
+
 	useEffect(() => {
 		if (skip !== null && skip !== undefined) {
-			listListeners(skip, limit, 0, null, null);
+			listListeners(skip, limit, 0, null, null, aliveOnly);
 			if (skip === 0) setPrevDisabled(true);
 			else setPrevDisabled(false);
 		}
@@ -123,11 +135,18 @@ export function Listeners(props: ListenerProps) {
 		} else {
 			// set the interval to fetch the job's log
 			const interval = setInterval(() => {
-				listListeners(skip, limit, 0, selectedCategory, selectedLabel);
+				listListeners(
+					skip,
+					limit,
+					0,
+					selectedCategory,
+					selectedLabel,
+					aliveOnly
+				);
 			}, config.extractorLivelihoodInterval);
 			return () => clearInterval(interval);
 		}
-	}, [searchText, listeners, skip, selectedCategory, selectedLabel]);
+	}, [searchText, listeners, skip, selectedCategory, selectedLabel, aliveOnly]);
 
 	// for pagination keep flipping until the return dataset is less than the limit
 	const previous = () => {
@@ -152,14 +171,28 @@ export function Listeners(props: ListenerProps) {
 		const selectedCategoryValue = (event.target as HTMLInputElement).value;
 		setSelectedCategory(selectedCategoryValue);
 		setSearchText("");
-		listListeners(skip, limit, 0, selectedCategoryValue, selectedLabel);
+		listListeners(
+			skip,
+			limit,
+			0,
+			selectedCategoryValue,
+			selectedLabel,
+			aliveOnly
+		);
 	};
 
 	const handleLabelChange = (event: React.ChangeEvent<HTMLInputElement>) => {
 		const selectedLabelValue = (event.target as HTMLInputElement).value;
 		setSelectedLabel(selectedLabelValue);
 		setSearchText("");
-		listListeners(skip, limit, 0, selectedCategory, selectedLabelValue);
+		listListeners(
+			skip,
+			limit,
+			0,
+			selectedCategory,
+			selectedLabelValue,
+			aliveOnly
+		);
 	};
 
 	const handleSubmitExtractionClose = () => {
@@ -189,7 +222,7 @@ export function Listeners(props: ListenerProps) {
 				spacing={3}
 			>
 				{/*categories*/}
-				<Grid item xs={6}>
+				<Grid item xs={5}>
 					<FormControl variant="standard" sx={{ width: "100%" }}>
 						<InputLabel id="label-categories">Filter by category</InputLabel>
 						<Select
@@ -209,7 +242,7 @@ export function Listeners(props: ListenerProps) {
 						</Select>
 					</FormControl>
 				</Grid>
-				<Grid item xs={6}>
+				<Grid item xs={5}>
 					<FormControl variant="standard" sx={{ width: "100%" }}>
 						<InputLabel id="label-categories">Filter by labels</InputLabel>
 						<Select
@@ -227,6 +260,21 @@ export function Listeners(props: ListenerProps) {
 						</Select>
 					</FormControl>
 				</Grid>
+				<Grid item xs={2}>
+					<FormControlLabel
+						sx={{ margin: "auto auto -2.5em auto" }}
+						control={
+							<Switch
+								color="primary"
+								checked={aliveOnly}
+								onChange={() => {
+									setAliveOnly(!aliveOnly);
+								}}
+							/>
+						}
+						label="Alive Extractors"
+					/>
+				</Grid>
 			</Grid>
 			<Grid container>
 				<Grid item xs={12}>
diff --git a/frontend/src/openapi/v2/services/ListenersService.ts b/frontend/src/openapi/v2/services/ListenersService.ts
@@ -27,13 +27,16 @@ export class ListenersService {
      * Arguments:
      * skip -- number of initial records to skip (i.e. for pagination)
      * limit -- restrict number of records to be returned (i.e. for pagination)
+     * heartbeat_interval -- number of seconds after which a listener is considered dead
      * category -- filter by category has to be exact match
      * label -- filter by label has to be exact match
+     * alive_only -- filter by alive status
      * @param skip
      * @param limit
      * @param heartbeatInterval
      * @param category
      * @param label
+     * @param aliveOnly
      * @returns EventListenerOut Successful Response
      * @throws ApiError
      */
@@ -43,6 +46,7 @@ export class ListenersService {
         heartbeatInterval: number = 300,
         category?: string,
         label?: string,
+        aliveOnly: boolean = false,
     ): CancelablePromise<Array<EventListenerOut>> {
         return __request({
             method: 'GET',
@@ -53,6 +57,7 @@ export class ListenersService {
                 'heartbeat_interval': heartbeatInterval,
                 'category': category,
                 'label': label,
+                'alive_only': aliveOnly,
             },
             errors: {
                 422: `Validation Error`,
