769 leveraging the heartbeat info to show extractor status (#822)

* add version chips

* update heart beat for keep the alive time

* update heartbeat

* update endpoints for livelihood

* add test

* black reformat

* codegen

* optional set to none

* frontend logic wired in

* I can't do math...

* try utcnow

* the logic of disable is correct now

* add version chip to pop up as well

* disable the description

* take out the extractors at root

* reformat

* update the symbol for extractor livelihood

* capitalize O

* update the indicator

* set time interval to pull etractor status every minute

* reformat heartbeat listener lint

* Updated comments.

---------

Co-authored-by: Luigi Marini <[email protected]>

Author: longshuicy

backend/app/config.py

@@ -79,5 +79,8 @@ class Settings(BaseSettings):
     RABBITMQ_HOST: str = "127.0.0.1"
     HEARTBEAT_EXCHANGE: str = "extractors"
 
+    # defautl listener heartbeat time interval in seconds 5 minutes
+    listener_heartbeat_interval = 5 * 60
+
 
 settings = Settings()

backend/app/models/listeners.py

@@ -69,6 +69,8 @@ class EventListenerDB(Document, EventListenerBase):
     creator: Optional[UserOut] = None
     created: datetime = Field(default_factory=datetime.now)
     modified: datetime = Field(default_factory=datetime.now)
+    lastAlive: datetime = None
+    alive: Optional[bool] = None  # made up field to indicate if extractor is alive
     properties: Optional[ExtractorInfo] = None
 
     class Settings:

backend/app/routers/listeners.py

@@ -74,6 +74,23 @@ async def _process_incoming_v1_extractor_info(
         return new_feed
 
 
+async def _check_livelihood(
+    listener: EventListenerDB, heartbeat_interval=settings.listener_heartbeat_interval
+):
+    if heartbeat_interval == 0:
+        heartbeat_interval = settings.listener_heartbeat_interval
+
+    if listener.lastAlive is None:
+        return False
+
+    now = datetime.datetime.utcnow()
+    elapsed = now - listener.lastAlive
+    if elapsed.total_seconds() > heartbeat_interval:
+        return False
+    else:
+        return True
+
+
 @router.get("/instance")
 async def get_instance_id(
     user=Depends(get_current_user),
@@ -150,7 +167,11 @@ async def save_legacy_listener(
 
 @router.get("/search", response_model=List[EventListenerOut])
 async def search_listeners(
-    text: str = "", skip: int = 0, limit: int = 2, user=Depends(get_current_username)
+    text: str = "",
+    skip: int = 0,
+    limit: int = 2,
+    heartbeat_interval: Optional[int] = settings.listener_heartbeat_interval,
+    user=Depends(get_current_username),
 ):
     """Search all Event Listeners in the db based on text.
 
@@ -171,7 +192,14 @@ async def search_listeners(
         .limit(limit)
         .to_list()
     )
-    return [listener.dict() for listener in listeners]
+
+    # batch return listener statuses for easy consumption
+    listenerResponse = []
+    for listener in listeners:
+        listener.alive = await _check_livelihood(listener, heartbeat_interval)
+        listenerResponse.append(listener.dict())
+
+    return listenerResponse
 
 
 @router.get("/categories", response_model=List[str])
@@ -196,11 +224,26 @@ async def get_listener(listener_id: str, user=Depends(get_current_username)):
     raise HTTPException(status_code=404, detail=f"listener {listener_id} not found")
 
 
+@router.get("/{listener_id}/status", response_model=bool)
+async def check_listener_livelihood(
+    listener_id: str,
+    heartbeat_interval: Optional[int] = settings.listener_heartbeat_interval,
+    user=Depends(get_current_username),
+):
+    """Return JSON information about an Event Listener if it exists."""
+    if (
+        listener := await EventListenerDB.get(PydanticObjectId(listener_id))
+    ) is not None:
+        return await _check_livelihood(listener, heartbeat_interval)
+    raise HTTPException(status_code=404, detail=f"listener {listener_id} not found")
+
+
 @router.get("", response_model=List[EventListenerOut])
 async def get_listeners(
     user_id=Depends(get_current_username),
     skip: int = 0,
     limit: int = 2,
+    heartbeat_interval: Optional[int] = settings.listener_heartbeat_interval,
     category: Optional[str] = None,
     label: Optional[str] = None,
 ):
@@ -219,7 +262,14 @@ async def get_listeners(
         query.append(EventListenerDB.properties.default_labels == label)
 
     listeners = await EventListenerDB.find(*query, skip=skip, limit=limit).to_list()
-    return [listener.dict() for listener in listeners]
+
+    # batch return listener statuses for easy consumption
+    listenerResponse = []
+    for listener in listeners:
+        listener.alive = await _check_livelihood(listener, heartbeat_interval)
+        listenerResponse.append(listener.dict())
+
+    return listenerResponse
 
 
 @router.put("/{listener_id}", response_model=EventListenerOut)

backend/app/tests/test_extractors.py

@@ -1,5 +1,5 @@
-import os
 from fastapi.testclient import TestClient
+
 from app.config import settings
 from app.tests.utils import create_dataset, upload_file, register_v1_extractor
 
@@ -18,6 +18,15 @@ def test_get_one(client: TestClient, headers: dict):
     assert response.json().get("id") is not None
 
 
+def test_get_status(client: TestClient, headers: dict):
+    extractor_id = register_v1_extractor(client, headers).get("id")
+    response = client.get(
+        f"{settings.API_V2_STR}/listeners/{extractor_id}/status", headers=headers
+    )
+    assert response.status_code == 200
+    assert response.json() is False
+
+
 def test_delete(client: TestClient, headers: dict):
     ext_name = "test.test_delete"
     extractor_id = register_v1_extractor(client, headers, ext_name).get("id")

backend/heartbeat_listener.py

@@ -2,6 +2,8 @@
 import json
 import logging
 import os
+from datetime import datetime
+
 from aio_pika import connect_robust
 from aio_pika.abc import AbstractIncomingMessage
 from packaging import version
@@ -30,27 +32,35 @@ async def callback(message: AbstractIncomingMessage):
             **extractor_info, properties=ExtractorInfo(**extractor_info)
         )
 
-        # check to see if extractor alredy exists and update if so
+        # check to see if extractor already exists and update if so
         existing_extractor = await EventListenerDB.find_one(
             EventListenerDB.name == msg["queue"]
         )
         if existing_extractor is not None:
-            # Update existing listener
+            extractor_db.id = existing_extractor.id
+            extractor_db.created = existing_extractor.created
+
+            # Update existing listener version
             existing_version = existing_extractor.version
             new_version = extractor_db.version
             if version.parse(new_version) > version.parse(existing_version):
-                # if this is a new version, add it to the database
-                extractor_db.id = existing_extractor.id
-                extractor_db.created = existing_extractor.created
-                new_extractor = await extractor_db.replace()
-                extractor_out = EventListenerOut(**new_extractor.dict())
                 logger.info(
                     "%s updated from %s to %s"
                     % (extractor_name, existing_version, new_version)
                 )
-                return extractor_out
+
+            extractor_db.lastAlive = datetime.utcnow()
+            logger.info("%s is alive at %s" % (extractor_name, str(datetime.utcnow())))
+            # Update existing listeners alive status
+            new_extractor = await extractor_db.replace()
+            extractor_out = EventListenerOut(**new_extractor.dict())
+
+            return extractor_out
+
         else:
             # Register new listener
+            extractor_db.lastAlive = datetime.utcnow()
+            logger.info("%s is alive at %s" % (extractor_name, str(datetime.utcnow())))
             new_extractor = await extractor_db.insert()
             extractor_out = EventListenerOut(**new_extractor.dict())
             logger.info("New extractor registered: " + extractor_name)

frontend/src/actions/listeners.js

@@ -6,6 +6,7 @@ export const RECEIVE_LISTENERS = "RECEIVE_LISTENERS";
 export function fetchListeners(
 	skip = 0,
 	limit = 21,
+	heartbeatInterval = 0,
 	category = null,
 	label = null
 ) {
@@ -14,6 +15,7 @@ export function fetchListeners(
 		return V2.ListenersService.getListenersApiV2ListenersGet(
 			skip,
 			limit,
+			heartbeatInterval,
 			category,
 			label
 		)
@@ -26,21 +28,30 @@ export function fetchListeners(
 			})
 			.catch((reason) => {
 				dispatch(
-					handleErrors(reason, fetchListeners(skip, limit, category, label))
+					handleErrors(
+						reason,
+						fetchListeners(skip, limit, heartbeatInterval, category, label)
+					)
 				);
 			});
 	};
 }
 
 export const SEARCH_LISTENERS = "SEARCH_LISTENERS";
 
-export function queryListeners(text, skip = 0, limit = 21) {
+export function queryListeners(
+	text,
+	skip = 0,
+	limit = 21,
+	heartbeatInterval = 0
+) {
 	return (dispatch) => {
 		// TODO: Parameters for dates? paging?
 		return V2.ListenersService.searchListenersApiV2ListenersSearchGet(
 			text,
 			skip,
-			limit
+			limit,
+			heartbeatInterval
 		)
 			.then((json) => {
 				dispatch({
@@ -50,7 +61,12 @@ export function queryListeners(text, skip = 0, limit = 21) {
 				});
 			})
 			.catch((reason) => {
-				dispatch(handleErrors(reason, queryListeners(text, skip, limit)));
+				dispatch(
+					handleErrors(
+						reason,
+						queryListeners(text, skip, limit, heartbeatInterval)
+					)
+				);
 			});
 	};
 }

frontend/src/app.config.ts

@@ -12,7 +12,8 @@ interface Config {
 	KeycloakRegister: string;
 	searchEndpoint: string;
 	refreshTokenInterval: number;
-	extractorInterval: number;
+	extractorStatusInterval: number;
+	extractorLivelihoodInterval: number;
 	eventListenerJobStatus: EventListenerJobStatus;
 	streamingBytes: number;
 	rawDataVisualizationThreshold: number;
@@ -36,18 +37,21 @@ config["GHIssueBaseURL"] =
 
 // Backend Keycloak login url
 config["KeycloakBaseURL"] =
-	process.env.KeycloakBaseURL || config.hostname + "/api/v2/auth";
-config["KeycloakLogin"] = config.KeycloakBaseURL + "/login";
-config["KeycloakLogout"] = config.KeycloakBaseURL + "/logout";
-config["KeycloakRefresh"] = config.KeycloakBaseURL + "/refresh_token";
-config["KeycloakRegister"] = config.KeycloakBaseURL + "/register";
+	process.env.KeycloakBaseURL || `${config.hostname}/api/v2/auth`;
+config["KeycloakLogin"] = `${config.KeycloakBaseURL}/login`;
+config["KeycloakLogout"] = `${config.KeycloakBaseURL}/logout`;
+config["KeycloakRefresh"] = `${config.KeycloakBaseURL}/refresh_token`;
+config["KeycloakRegister"] = `${config.KeycloakBaseURL}/register`;
 
 // elasticsearch
 config["searchEndpoint"] = `${hostname}/api/v2/elasticsearch`;
 
-// refresh toekn time interval
-config["refreshTokenInterval"] = 1000 * 60; // milliseconds
-config["extractorInterval"] = 2000; // milliseconds
+// refresh token time interval
+config["refreshTokenInterval"] = 1000 * 60; // 1 minute
+// updated extractor logs
+config["extractorStatusInterval"] = 1000 * 10; // 10 seconds
+// update extractor stutus (offline/online)
+config["extractorLivelihoodInterval"] = 1000 * 120; // 2 minutes
 
 config["eventListenerJobStatus"] = <EventListenerJobStatus>{};
 config["eventListenerJobStatus"]["created"] = "CREATED";

frontend/src/components/Explore.tsx

@@ -9,8 +9,7 @@ import { a11yProps, TabPanel } from "./tabs/TabComponent";
 import DatasetCard from "./datasets/DatasetCard";
 import { ArrowBack, ArrowForward } from "@material-ui/icons";
 import Layout from "./Layout";
-import { Link as RouterLink, useLocation } from "react-router-dom";
-import { Listeners } from "./listeners/Listeners";
+import { Link as RouterLink } from "react-router-dom";
 import { ErrorModal } from "./errors/ErrorModal";
 
 const tab = {
@@ -96,12 +95,11 @@ export const Explore = (): JSX.Element => {
 							aria-label="dashboard tabs"
 						>
 							<Tab sx={tab} label="Datasets" {...a11yProps(0)} />
-							<Tab sx={tab} label="Extractors" {...a11yProps(1)} />
 						</Tabs>
 					</Box>
 					<TabPanel value={selectedTabIndex} index={0}>
 						<Grid container spacing={2}>
-							{datasets !== undefined ?  (
+							{datasets !== undefined ? (
 								datasets.map((dataset) => {
 									return (
 										<Grid item key={dataset.id} xs={12} sm={6} md={4} lg={3}>
@@ -122,12 +120,17 @@ export const Explore = (): JSX.Element => {
 							{datasets.length === 0 ? (
 								<Grid container justifyContent="center">
 									<Box textAlign="center">
-										<p>Nobody has created any datasets on this instance. Click below to create a dataset!</p>
-										<Button component={RouterLink} to="/create-dataset"
+										<p>
+											Nobody has created any datasets on this instance. Click
+											below to create a dataset!
+										</p>
+										<Button
+											component={RouterLink}
+											to="/create-dataset"
 											variant="contained"
 											sx={{ m: 2 }}
 										>
-										Create Dataset
+											Create Dataset
 										</Button>
 									</Box>
 								</Grid>
@@ -157,13 +160,10 @@ export const Explore = (): JSX.Element => {
 									</Button>
 								</ButtonGroup>
 							</Box>
-						): (
+						) : (
 							<></>
 						)}
 					</TabPanel>
-					<TabPanel value={selectedTabIndex} index={1}>
-						<Listeners />
-					</TabPanel>
 					<TabPanel value={selectedTabIndex} index={4} />
 					<TabPanel value={selectedTabIndex} index={2} />
 					<TabPanel value={selectedTabIndex} index={3} />

frontend/src/components/listeners/ExtractionHistory.tsx

@@ -77,9 +77,19 @@ export const ExtractionHistory = (): JSX.Element => {
 	const listListeners = (
 		skip: number | undefined,
 		limit: number | undefined,
+		heartbeatInterval: number | undefined,
 		selectedCategory: string | null,
 		selectedLabel: string | null
-	) => dispatch(fetchListeners(skip, limit, selectedCategory, selectedLabel));
+	) =>
+		dispatch(
+			fetchListeners(
+				skip,
+				limit,
+				heartbeatInterval,
+				selectedCategory,
+				selectedLabel
+			)
+		);
 	const listListenerJobs = (
 		listenerId: string | null,
 		status: string | null,
@@ -118,7 +128,7 @@ export const ExtractionHistory = (): JSX.Element => {
 	const [selectedCreatedTime, setSelectedCreatedTime] = useState(null);
 
 	useEffect(() => {
-		listListeners(skip, limit, null, null);
+		listListeners(skip, limit, 0, null, null);
 		listListenerJobs(null, null, null, null, null, null, 0, 100);
 	}, []);
 
@@ -180,7 +190,7 @@ export const ExtractionHistory = (): JSX.Element => {
 
 	useEffect(() => {
 		if (skip !== null && skip !== undefined) {
-			listListeners(skip, limit, null, null);
+			listListeners(skip, limit, 0, null, null);
 			if (skip === 0) setPrevDisabled(true);
 			else setPrevDisabled(false);
 		}