marker/shuffle-cracked
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sending tenzir health check status to backend

Browse Source
This commit is contained in:
Hari Krishna
2024-06-23 09:13:08 +00:00
committed by satti-hari-krishna-reddy
parent ad92e5e722
commit ab3c755ad3
2 changed files with 90 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/go-app/main.go
+41
View File
@@ -2149,6 +2149,46 @@ func parseConcatenatedJSONLogs(logs string) ([]map[string]interface{}, error) {
return jsonList, nil
}
func handleTenzirHealthUpdate(resp http.ResponseWriter, request *http.Request) {
if request.Method != "POST" {
request.Method = "POST"
}
type HealthUpdate struct {
Status string `json:"status"`
}
var healthUpdate HealthUpdate
err := json.NewDecoder(request.Body).Decode(&healthUpdate)
if err != nil {
resp.WriteHeader(http.StatusBadRequest)
fmt.Fprintf(resp, "Failed to decode JSON: %v", err)
return
}
ctx := context.Background()
status := healthUpdate.Status
result, err := shuffle.GetDisabledRules(ctx)
if (err != nil && err.Error() != "rules doesn't exist") || err == nil {
result.IsTenzirActive = status
result.LastActive = time.Now().Unix()
err = shuffle.StoreDisabledRules(ctx, *result)
if err != nil {
resp.WriteHeader(500)
resp.Write([]byte(`{"success": false}`))
return
}
resp.WriteHeader(200)
resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
return
}
resp.WriteHeader(500)
resp.Write([]byte(`{"success": false}`))
return
}
func executeCloudAction(action shuffle.CloudSyncJob, apikey string) error {
data, err := json.Marshal(action)
if err != nil {
@@ -5091,6 +5131,7 @@ func initHandlers() {
r.HandleFunc("/api/v1/triggers/pipeline", shuffle.HandleNewPipelineRegister).Methods("POST", "OPTIONS")
//r.HandleFunc("/api/v1/triggers/pipeline/save", shuffle.HandleSavePipelineInfo).Methods("PUT", "OPTIONS")
r.HandleFunc("/api/v1/pipelines/{key}", handlePipelineCallback).Methods("POST", "GET", "PATCH", "PUT", "DELETE", "OPTIONS")
r.HandleFunc("/api/v1/pipelines/tenzir_node_health", handleTenzirHealthUpdate).Methods("POST","OPTIONS")
r.HandleFunc("/api/v1/triggers", shuffle.HandleGetTriggers).Methods("GET", "OPTIONS")
//r.HandleFunc("/api/v1/triggers/gmail/routing", handleGmailRouting).Methods("POST", "OPTIONS")
functions/onprem/orborus/orborus.go
+49 -49
View File
@@ -1806,6 +1806,12 @@ func main() {
log.Printf("[WARNING] Defaulting to environment name %s. Set environment variable ENVIRONMENT_NAME to change. This should be the same as in the frontend action.", environment)
}
if tenzirUrl == "" {
tenzirUrl = "http://localhost:5160"
log.Printf("[WARNING] SHUFFLE_TENZIR_URL not set, falling back to default URL: %s",tenzirUrl)
}
// FIXME - during init, BUILD and/or LOAD worker and app_sdk
// Build/load app_sdk so it can be loaded as 127.0.0.1:5000/walkoff_app_sdk
log.Printf("[INFO] Setting up Docker environment. Downloading worker and App SDK!")
@@ -2254,7 +2260,7 @@ func main() {
}
}
_ = sendTenzirHealthStatus()
time.Sleep(time.Duration(sleepTime) * time.Second)
}
}
@@ -2414,11 +2420,6 @@ func main() {
// docker run tenzir/tenzir:latest 'from http://192.168.86.44:5002/api/v1/orgs/7e9b9007-5df2-4b47-bca5-c4d267ef2943/cache/CIDR%20ranges?type=text&authorization=cec9d01f-09b2-4419-8a0a-76c6046e3fef read lines | to http://192.168.86.44:5002/api/v1/hooks/webhook_665ace5f-f27b-496a-a365-6e07eb61078c write lines'
func handlePipeline(incRequest shuffle.ExecutionRequest) error {
if tenzirUrl == "" {
tenzirUrl = "http://localhost:5160"
log.Printf("[WARNING] SHUFFLE_TENZIR_URL not set, falling back to default URL: %s", tenzirUrl)
}
err := deployTenzirNode()
if err != nil {
log.Printf("[ERROR] failed to deploy the pipeline, reason: %s", err)
@@ -3072,53 +3073,52 @@ func removePath(containerName, path string) error {
return nil
}
// func savePipelineData(pipelineId, identifier, status string) error {
func sendTenzirHealthStatus() error {
var status string
url := fmt.Sprintf("%s/api/v1/triggers/pipeline/tenzir_node_health", baseUrl)
err := checkTenzirNode()
if err != nil {
return err
} else {
status = "active"
}
// url := fmt.Sprintf("%s/api/v1/triggers/pipeline/save", baseUrl)
// identifierWithoutPrefix := strings.TrimPrefix(identifier, "shuffle-")
forwardMethod := "POST"
payload := map[string]interface{}{
"status": status,
}
payloadBytes, err := json.Marshal(payload)
if err != nil {
log.Printf("[ERROR] Failed to marshal payload: %s", err)
return err
}
forwardData := bytes.NewBuffer(payloadBytes)
req, err := http.NewRequest(
forwardMethod,
url,
forwardData,
)
if err != nil {
log.Printf("[ERROR] Failed to create HTTP request: %s", err)
return err
}
req.Header.Set("Content-Type", "application/json")
// forwardMethod := "PUT"
client := &http.Client{Timeout: 10 * time.Second}
resp, err := client.Do(req)
if err != nil {
log.Printf("[ERROR] Failed to send HTTP request: %s", err)
return err
}
defer resp.Body.Close()
// payload := map[string]interface{}{
// "pipeline_id": pipelineId,
// "trigger_id": identifierWithoutPrefix,
// "status": status,
// }
if resp.StatusCode != 200 {
log.Printf("[ERROR] Received non-successful HTTP status code: %d", resp.StatusCode)
return fmt.Errorf("unexpected HTTP status code: %d", resp.StatusCode)
}
// payloadBytes, err := json.Marshal(payload)
// if err != nil {
// log.Printf("[ERROR] Failed to marshal payload: %s", err)
// return err
// }
// forwardData := bytes.NewBuffer(payloadBytes)
// req, err := http.NewRequest(
// forwardMethod,
// url,
// forwardData,
// )
// if err != nil {
// log.Printf("[ERROR] Failed to create HTTP request: %s", err)
// return err
// }
// req.Header.Set("Content-Type", "application/json")
// client := &http.Client{Timeout: 10 * time.Second}
// resp, err := client.Do(req)
// if err != nil {
// log.Printf("[ERROR] Failed to send HTTP request: %s", err)
// return err
// }
// defer resp.Body.Close()
// if resp.StatusCode != 200 {
// log.Printf("[ERROR] Received non-successful HTTP status code: %d", resp.StatusCode)
// return fmt.Errorf("unexpected HTTP status code: %d", resp.StatusCode)
// }
// return nil
// }
return nil
}
// Is this ok to do with Docker? idk :)
func getRunningWorkers(ctx context.Context, workerTimeout int) int {