runner: enhance monitoring handler for airflow

Author: wuriyanto48

dashboard/ui/src/pages/dashboard.js

@@ -52,8 +52,9 @@ function Dashboard() {
 
           const statusPriority = {
             DOWN: 0,
-            MONITORED: 1,
-            UP: 2,
+            CHECKING: 1,
+            MONITORED: 2,
+            UP: 3,
           };
 
           serviceArray.sort((a, b) => {
@@ -121,15 +122,28 @@ function Dashboard() {
       backgroundColor:
         status === 'UP' ? '#28a745' :
         status === 'DOWN' ? '#dc3545' :
-        '#ffc107', 
+        status === 'MONITORED' ? '#ffc107' : 
+        '#b4e83a', 
       whiteSpace: 'nowrap',
       animation: `pulse ${randomDuration.toFixed(2)}s infinite`,
       flexShrink: 0,
     };
   };
-  
-  
-  
+
+  const getMessageDetailsStyle = (service) => {
+    return {
+      fontSize: '14px',
+      color:
+        service.status === 'UP' ? '#28a745' :
+        service.status === 'DOWN' ? '#dc3545' :
+        service.status === 'MONITORED' ? '#ffc107' :
+        '#b4e83a',
+      marginBottom: '10px',
+      wordWrap: 'break-word',
+      overflowWrap: 'break-word',
+      whiteSpace: 'pre-line',
+    };
+  };
 
   const getTagsStyle = () => ({
     display: 'flex',
@@ -277,6 +291,8 @@ function Dashboard() {
                   ? 'Not OK'
                   : service.status === 'MONITORED'
                   ? 'Monitored'
+                  : service.status == 'CHECKING'
+                  ? 'Checking'
                   : service.status}
               </span>
             </div>
@@ -352,14 +368,7 @@ function Dashboard() {
             )
             :
             (
-              <div style={{ 
-                fontSize: '14px', 
-                color: '#dc3545', 
-                marginBottom: '10px', 
-                wordWrap: 'break-word',
-                overflowWrap: 'break-word',
-                whiteSpace: 'pre-line',
-              }}>
+              <div style={getMessageDetailsStyle(service)}>
                 {service.messageDetails}
               </div>
             )

runner/runner.go

@@ -227,6 +227,29 @@ func healthCheck(n string, s tob.Service, t *time.Ticker, waiter tob.Waiter) {
 			resp := s.Ping()
 			respStr := string(resp)
 
+			// Airflow Monitoring
+			if s.Name() == string(tob.Airflow) {
+				for _, notificator := range s.GetNotificators() {
+					if !util.IsNilish(notificator) {
+						if notificator.IsEnabled() && notificator.Provider() == "webhook" {
+							notificatorMessage := fmt.Sprintf("%s is DOWN", n)
+							if s.GetMessage() != "" {
+								notificatorMessage = fmt.Sprintf("%s is CHECKING | %s", n, s.GetMessage())
+								if respStr == tob.NotOk {
+									notificatorMessage = fmt.Sprintf("%s is DOWN | %s", n, s.GetMessage())
+								}
+							}
+							if notificator.IsEnabled() && s.Name() != string(tob.SSLStatus) {
+								err := notificator.Send(notificatorMessage)
+								if err != nil {
+									tob.Logger.Printf("notificator %s error: %s", notificator.Provider(), err.Error())
+								}
+							}
+						}
+					}
+				}
+			}
+
 			// SSL Monitoring
 			if s.Name() == string(tob.SSLStatus) {
 				for _, notificator := range s.GetNotificators() {
@@ -275,7 +298,7 @@ func healthCheck(n string, s tob.Service, t *time.Ticker, waiter tob.Waiter) {
 
 				notificatorMessage := fmt.Sprintf("%s is UP. It was down for %s", n, s.GetDownTimeDiff())
 				if s.GetMessage() != "" {
-					notificatorMessage = fmt.Sprintf("%s %s", n, s.GetMessage())
+					notificatorMessage = fmt.Sprintf("%s is UP | %s", n, s.GetMessage())
 				}
 
 				for _, notificator := range s.GetNotificators() {

services/airflow/airflow.go

@@ -17,18 +17,19 @@ import (
 
 // Airflow service
 type Airflow struct {
-	url                string
-	recovered          bool
-	lastDownTime       string
-	schedulerStatus    string
-	metadatabaseStatus string
-	enabled            bool
-	verbose            bool
-	logger             *log.Logger
-	checkInterval      int
-	stopChan           chan bool
-	message            string
-	notificatorConfig  config.Config
+	url                      string
+	recovered                bool
+	lastDownTime             string
+	schedulerStatus          string
+	latestSchedulerHeartbeat string
+	metadatabaseStatus       string
+	enabled                  bool
+	verbose                  bool
+	logger                   *log.Logger
+	checkInterval            int
+	stopChan                 chan bool
+	message                  string
+	notificatorConfig        config.Config
 }
 
 // Airflow's constructor
@@ -72,35 +73,86 @@ func (a *Airflow) checkClusterStatus(resp *http.Response) error {
 		return err
 	}
 
-	schedulerStatus, ok := data["scheduler"].(map[string]interface{})["status"].(string)
+	schedulerRaw, ok := data["scheduler"].(map[string]interface{})
 	if !ok {
 		if a.verbose {
-			a.logger.Printf("cannot read scheduler status: %v\n", err)
+			a.logger.Println("cannot read scheduler block (not a map)")
 		}
+		return errors.New("invalid scheduler block")
+	}
 
-		return err
+	schedulerStatus, ok := schedulerRaw["status"].(string)
+	if !ok {
+		if a.verbose {
+			a.logger.Println("cannot read scheduler status (not a string)")
+		}
+
+		return errors.New("invalid scheduler status")
+	}
+
+	latestSchedulerHeartbeat, ok := schedulerRaw["latest_scheduler_heartbeat"].(string)
+	if !ok {
+		if a.verbose {
+			a.logger.Println("cannot read scheduler latest_scheduler_heartbeat (not a string)")
+		}
+
+		return errors.New("invalid latest_scheduler_heartbeat")
+	}
+
+	utcTime, err := time.Parse(time.RFC3339Nano, latestSchedulerHeartbeat)
+	if err != nil {
+		if a.verbose {
+			a.logger.Printf("failed to parse time: %v\n", err)
+		}
+
+		return errors.New("invalid latest_scheduler_heartbeat")
 	}
+
+	timezoneJakarta, err := time.LoadLocation("Asia/Jakarta")
+	if err != nil {
+		if a.verbose {
+			a.logger.Printf("failed to load WIB location: %v\n", err)
+		}
+
+		return errors.New("invalid timezone")
+	}
+
+	wibTime := utcTime.In(timezoneJakarta)
+	a.latestSchedulerHeartbeat = wibTime.String()
+
 	a.schedulerStatus = schedulerStatus
 
-	metadatabaseStatus, ok := data["metadatabase"].(map[string]interface{})["status"].(string)
+	metadatabaseRaw, ok := data["metadatabase"].(map[string]interface{})
 	if !ok {
 		if a.verbose {
-			a.logger.Printf("cannot read metadatabase status: %v\n", err)
+			a.logger.Println("cannot read metadatabase block (not a map)")
 		}
+		return errors.New("invalid metadatabase block")
+	}
 
-		return err
+	metadatabaseStatus, ok := metadatabaseRaw["status"].(string)
+	if !ok {
+		if a.verbose {
+			a.logger.Println("cannot read metadatabase status")
+		}
+
+		return errors.New("invalid metadatabase status")
 	}
+
 	a.metadatabaseStatus = metadatabaseStatus
 
+	message := fmt.Sprintf("Airflow Scheduler Status: %s\nLatest Scheduler Heartbeat: %s\n\n\n Airflow Metadatabase: %s",
+		a.schedulerStatus, a.latestSchedulerHeartbeat, a.metadatabaseStatus)
 	if a.schedulerStatus != "healthy" || a.metadatabaseStatus != "healthy" {
-		errMsg := fmt.Sprintf("airflow is unhealthy: scheduler (%s), metadatabase (%s)", a.schedulerStatus, a.metadatabaseStatus)
 		if a.verbose {
-			a.logger.Println(errMsg)
+			a.logger.Println(message)
 		}
 
-		return errors.New(errMsg)
+		return errors.New(message)
 	}
 
+	a.SetMessage(message)
+
 	return nil
 }