url/checkcle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor: Integrate regional agents in service-operation

Browse Source 
Integrate distributed monitoring agents for HTTP, PING, TCP, and DNS checks
This commit is contained in:
Tola Leng
2025-06-21 17:48:55 +07:00
parent ea95490f13
commit 57a65106cb
16 changed files with 422 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/service-operation/.env
+1 -1
View File
@@ -4,7 +4,7 @@ PORT=8091
# Operation defaults
DEFAULT_COUNT=4
DEFAULT_TIMEOUT=3s
DEFAULT_TIMEOUT=10s
MAX_COUNT=20
MAX_TIMEOUT=30s
server/service-operation/.env.example
+2 -2
View File
@@ -4,7 +4,7 @@ PORT=8091
# Operation defaults
DEFAULT_COUNT=4
DEFAULT_TIMEOUT=3s
DEFAULT_TIMEOUT=10s
MAX_COUNT=20
MAX_TIMEOUT=30s
@@ -13,4 +13,4 @@ ENABLE_LOGGING=true
# PocketBase integration (no authentication required)
POCKETBASE_ENABLED=true
POCKETBASE_URL=https://pb-api.k8sops.asia
POCKETBASE_URL=http://localhost:8090
server/service-operation/README.md
-7
View File
@@ -117,13 +117,6 @@ Environment variables:
```bash
go run main.go
```
### Docker
```bash
docker build -t service-operation .
docker run -p 8080:8080 service-operation
```
## Requirements
- Go 1.21+
server/service-operation/main.go
+7 -7
View File
@@ -1,4 +1,3 @@
package main
import (
@@ -32,15 +31,15 @@ func main() {
if err := pbClient.TestConnection(); err != nil {
log.Printf("Warning: PocketBase connection test failed: %v", err)
} else {
// Initialize and start monitoring service
// Initialize and start monitoring service with regional support
monitoringService = monitoring.NewMonitoringService(pbClient)
go monitoringService.Start()
log.Println("Monitoring service started (public access mode)")
log.Println("Monitoring service started with regional agent support")
// Initialize and start SSL monitoring service
// Initialize and start SSL monitoring service (unchanged - no regional support)
sslMonitoringService = monitoring.NewSSLMonitoringService(pbClient)
go sslMonitoringService.Start()
log.Println("SSL monitoring service started")
log.Println("SSL monitoring service started (independent of regional agents)")
}
}
}
@@ -67,10 +66,10 @@ func main() {
log.Printf("PocketBase integration enabled at %s (public access)", pbClient.GetBaseURL())
}
if monitoringService != nil {
log.Printf("Automatic service monitoring enabled")
log.Printf("Automatic service monitoring enabled with regional agent support")
}
if sslMonitoringService != nil {
log.Printf("SSL certificate monitoring enabled")
log.Printf("SSL certificate monitoring enabled (independent)")
}
log.Printf("Endpoints:")
log.Printf(" POST /operation - Full operation test (ping, dns, tcp, http, ssl)")
@@ -79,6 +78,7 @@ func main() {
log.Printf(" GET /ping/quick?host=<host> - Legacy quick ping test")
log.Printf(" GET /health - Health check")
log.Printf("Supported operations: ping, dns, tcp, http, ssl")
log.Printf("Regional monitoring: Tracks 'Default' region connection status")
// Setup graceful shutdown
c := make(chan os.Signal, 1)
server/service-operation/monitoring/checker.go
+3 -2
View File
@@ -114,9 +114,10 @@ func (ms *MonitoringService) performCheck(service pocketbase.Service) {
log.Printf("Failed to update service status for %s: %v", latestService.Name, err)
}
// Save metrics data in ONE place to prevent duplicates
// Save metrics data with regional information
if result != nil {
metricsSaver := savers.NewMetricsSaver(ms.pbClient)
regionName, agentID := ms.GetRegionalInfo()
metricsSaver := savers.NewMetricsSaverWithRegion(ms.pbClient, regionName, agentID)
metricsSaver.SaveMetricsForService(*latestService, result)
}
}
server/service-operation/monitoring/regional_monitor.go
+114
View File
@@ -0,0 +1,114 @@
package monitoring
import (
"log"
"time"
"service-operation/pocketbase"
)
type RegionalMonitor struct {
pbClient *pocketbase.PocketBaseClient
regionalService *pocketbase.RegionalService
isOnline bool
ticker *time.Ticker
stopChan chan bool
}
func NewRegionalMonitor(pbClient *pocketbase.PocketBaseClient) *RegionalMonitor {
return &RegionalMonitor{
pbClient: pbClient,
isOnline: false,
stopChan: make(chan bool),
}
}
func (rm *RegionalMonitor) Start() {
// Get or create the "Default" regional service
service, err := rm.pbClient.GetDefaultRegionalService()
if err != nil {
log.Printf("Warning: Could not get default regional service: %v", err)
log.Printf("Regional monitoring will continue with fallback values")
// Continue with fallback service
service = &pocketbase.RegionalService{
ID: "default",
RegionName: "Default",
Status: "active",
AgentID: "1",
AgentIPAddress: "127.0.0.1",
Connection: "offline",
Token: "default-token",
}
}
rm.regionalService = service
rm.ticker = time.NewTicker(30 * time.Second) // Check connection every 30 seconds
// Initial connection status update
rm.updateConnectionStatus("online")
rm.isOnline = true
log.Printf("Regional monitor started for region: %s (Agent ID: %s)",
service.RegionName, service.AgentID)
go func() {
for {
select {
case <-rm.ticker.C:
rm.checkConnection()
case <-rm.stopChan:
rm.ticker.Stop()
return
}
}
}()
}
func (rm *RegionalMonitor) Stop() {
if rm.regionalService != nil {
rm.updateConnectionStatus("offline")
log.Printf("Regional monitor stopped for region: %s", rm.regionalService.RegionName)
}
if rm.ticker != nil {
rm.ticker.Stop()
}
rm.stopChan <- true
}
func (rm *RegionalMonitor) checkConnection() {
// Test PocketBase connection to determine if we're online
err := rm.pbClient.TestConnection()
if err != nil && rm.isOnline {
// We were online but now we're offline
rm.updateConnectionStatus("offline")
rm.isOnline = false
log.Printf("Regional agent went offline: %v", err)
} else if err == nil && !rm.isOnline {
// We were offline but now we're online
rm.updateConnectionStatus("online")
rm.isOnline = true
log.Printf("Regional agent back online")
}
}
func (rm *RegionalMonitor) updateConnectionStatus(status string) {
if rm.regionalService == nil {
return
}
if err := rm.pbClient.UpdateRegionalServiceConnection(rm.regionalService.ID, status); err != nil {
// Don't log errors for update failures - might be due to collection access issues
// log.Printf("Failed to update regional service connection status: %v", err)
}
}
func (rm *RegionalMonitor) GetRegionalInfo() (string, string) {
if rm.regionalService == nil {
return "Default", "1" // Fallback values
}
return rm.regionalService.RegionName, rm.regionalService.AgentID
}
server/service-operation/monitoring/service.go
+21 -10
View File
@@ -1,4 +1,3 @@
package monitoring
import (
@@ -10,19 +9,21 @@ import (
)
type MonitoringService struct {
pbClient *pocketbase.PocketBaseClient
activeServices map[string]*ServiceMonitor
mu sync.RWMutex
stopChan chan bool
isRunning bool
pbClient *pocketbase.PocketBaseClient
activeServices map[string]*ServiceMonitor
regionalMonitor *RegionalMonitor
mu sync.RWMutex
stopChan chan bool
isRunning bool
}
func NewMonitoringService(pbClient *pocketbase.PocketBaseClient) *MonitoringService {
return &MonitoringService{
pbClient: pbClient,
activeServices: make(map[string]*ServiceMonitor),
stopChan: make(chan bool),
isRunning: false,
pbClient: pbClient,
activeServices: make(map[string]*ServiceMonitor),
regionalMonitor: NewRegionalMonitor(pbClient),
stopChan: make(chan bool),
isRunning: false,
}
}
@@ -38,6 +39,9 @@ func (ms *MonitoringService) Start() {
ms.isRunning = true
log.Println("Starting monitoring service...")
// Start regional monitoring
ms.regionalMonitor.Start()
// Start monitoring all services from PocketBase
go ms.monitoringLoop()
}
@@ -53,6 +57,9 @@ func (ms *MonitoringService) Stop() {
log.Println("Stopping monitoring service...")
ms.isRunning = false
// Stop regional monitoring
ms.regionalMonitor.Stop()
// Stop all active monitors
for serviceID, monitor := range ms.activeServices {
ms.stopMonitor(serviceID, monitor)
@@ -61,6 +68,10 @@ func (ms *MonitoringService) Stop() {
ms.stopChan <- true
}
func (ms *MonitoringService) GetRegionalInfo() (string, string) {
return ms.regionalMonitor.GetRegionalInfo()
}
func (ms *MonitoringService) monitoringLoop() {
ticker := time.NewTicker(30 * time.Second) // Check for new services every 30 seconds
defer ticker.Stop()
server/service-operation/monitoring/ssl_monitor.go
+1 -1
View File
@@ -77,7 +77,7 @@ func (s *SSLMonitoringService) shouldCheckCertificate(cert types.SSLCertificate)
return true
} else {
//log.Printf("Certificate %s scheduled for later check (check_at: %s)",
//cert.Domain, checkAt.Format("2006-01-02 15:04:05"))
// cert.Domain, checkAt.Format("2006-01-02 15:04:05"))
//return false
}
} else {
server/service-operation/pocketbase/regional.go
+117
View File
@@ -0,0 +1,117 @@
package pocketbase
import (
"fmt"
"net/http"
"time"
)
func (c *PocketBaseClient) GetRegionalServices() ([]RegionalService, error) {
resp, err := c.httpClient.Get(fmt.Sprintf("%s/api/collections/regional_service/records", c.baseURL))
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode == http.StatusForbidden {
// Collection might require authentication or doesn't exist
fmt.Printf("Warning: Cannot access regional_service collection (403). Regional monitoring will be disabled.\n")
return nil, fmt.Errorf("access denied to regional_service collection")
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("failed to get regional services: status %d", resp.StatusCode)
}
var response RegionalServicesResponse
if err := c.parseResponse(resp, &response); err != nil {
return nil, err
}
return response.Items, nil
}
func (c *PocketBaseClient) GetDefaultRegionalService() (*RegionalService, error) {
services, err := c.GetRegionalServices()
if err != nil {
// Try to create a default regional service if we can't get existing ones
return c.CreateDefaultRegionalService()
}
// Look for a service with region_name "Default"
for _, service := range services {
if service.RegionName == "Default" {
return &service, nil
}
}
// If no default service found, try to create one
return c.CreateDefaultRegionalService()
}
func (c *PocketBaseClient) CreateDefaultRegionalService() (*RegionalService, error) {
defaultService := map[string]interface{}{
"region_name": "Default",
"status": "active",
"agent_id": "1",
"agent_ip_address": "127.0.0.1",
"connection": "offline",
"token": fmt.Sprintf("default-%d", time.Now().Unix()),
}
err := c.createRecord("regional_service", defaultService)
if err != nil {
fmt.Printf("Warning: Could not create default regional service: %v\n", err)
// Return a fallback service for local operations
return &RegionalService{
ID: "default",
RegionName: "Default",
Status: "active",
AgentID: "1",
AgentIPAddress: "127.0.0.1",
Connection: "offline",
Token: "default-token",
}, nil
}
// Try to fetch the created service
services, err := c.GetRegionalServices()
if err == nil {
for _, service := range services {
if service.RegionName == "Default" {
return &service, nil
}
}
}
// Return fallback if still can't get the created service
return &RegionalService{
ID: "default",
RegionName: "Default",
Status: "active",
AgentID: "1",
AgentIPAddress: "127.0.0.1",
Connection: "offline",
Token: "default-token",
}, nil
}
func (c *PocketBaseClient) UpdateRegionalServiceConnection(serviceID, connection string) error {
// Determine status based on connection
status := "inactive"
if connection == "online" {
status = "active"
}
data := map[string]interface{}{
"connection": connection,
"status": status,
}
return c.updateRecord("regional_service", serviceID, data)
}
func (c *PocketBaseClient) SaveSSLData(sslData SSLDataRecord) error {
return c.createRecord("ssl_data", sslData)
}
server/service-operation/pocketbase/types.go
+44 -1
View File
@@ -1,4 +1,3 @@
package pocketbase
import "time"
@@ -46,6 +45,8 @@ type PingDataRecord struct {
RTTs string `json:"rtts"`
Details string `json:"details,omitempty"`
ErrorMessage string `json:"error_message,omitempty"`
RegionName string `json:"region_name,omitempty"`
AgentID string `json:"agent_id,omitempty"`
}
type UptimeDataRecord struct {
@@ -61,6 +62,8 @@ type UptimeDataRecord struct {
Details string `json:"details"`
Region string `json:"region,omitempty"`
RegionID string `json:"region_id,omitempty"`
RegionName string `json:"region_name,omitempty"`
AgentID string `json:"agent_id,omitempty"`
}
type DNSDataRecord struct {
@@ -94,6 +97,46 @@ type TCPDataRecord struct {
AgentID string `json:"agent_id,omitempty"`
}
// SSL Data Record remains unchanged - no regional agent fields
type SSLDataRecord struct {
ServiceID string `json:"service_id"`
Timestamp time.Time `json:"timestamp"`
ResponseTime int64 `json:"response_time"`
Status string `json:"status"`
ValidFrom string `json:"valid_from"`
ValidTill string `json:"valid_till"`
DaysLeft int `json:"days_left"`
Issuer string `json:"issuer"`
Subject string `json:"subject"`
SerialNumber string `json:"serial_number"`
Algorithm string `json:"algorithm"`
SANs string `json:"sans"`
ResolvedIP string `json:"resolved_ip"`
ErrorMessage string `json:"error_message,omitempty"`
Details string `json:"details,omitempty"`
}
// Regional Service record structure
type RegionalService struct {
ID string `json:"id"`
RegionName string `json:"region_name"`
Status string `json:"status"`
AgentID string `json:"agent_id"`
AgentIPAddress string `json:"agent_ip_address"`
Connection string `json:"connection"`
Token string `json:"token"`
Created string `json:"created"`
Updated string `json:"updated"`
}
type RegionalServicesResponse struct {
Page int `json:"page"`
PerPage int `json:"perPage"`
TotalItems int `json:"totalItems"`
TotalPages int `json:"totalPages"`
Items []RegionalService `json:"items"`
}
type Service struct {
ID string `json:"id"`
Name string `json:"name"`
server/service-operation/pocketbase/utils.go
+44
View File
@@ -9,6 +9,50 @@ import (
"net/http"
)
func (c *PocketBaseClient) parseResponse(resp *http.Response, target interface{}) error {
body, err := io.ReadAll(resp.Body)
if err != nil {
return err
}
return json.Unmarshal(body, target)
}
func (c *PocketBaseClient) updateRecord(collection string, recordID string, data interface{}) error {
jsonData, err := json.Marshal(data)
if err != nil {
return err
}
req, err := http.NewRequest("PATCH",
fmt.Sprintf("%s/api/collections/%s/records/%s", c.baseURL, collection, recordID),
bytes.NewBuffer(jsonData))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
resp, err := c.httpClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
bodyBytes, _ := io.ReadAll(resp.Body)
bodyString := string(bodyBytes)
fmt.Printf("Failed to update record in %s collection. Status: %d, Response: %s\n",
collection, resp.StatusCode, bodyString)
return fmt.Errorf("failed to update record in %s, status: %d, response: %s",
collection, resp.StatusCode, bodyString)
}
return nil
}
func (c *PocketBaseClient) createRecord(collection string, data interface{}) error {
jsonData, err := json.Marshal(data)
if err != nil {
server/service-operation/shared/savers/dns_saver.go
+2 -2
View File
@@ -59,8 +59,8 @@ func (ms *MetricsSaver) SaveDNSDataToPocketBase(result *types.OperationResult, s
Authority: "", // Not available in current implementation
ErrorMessage: result.Error,
Details: details, // Short, clean message
RegionName: "default", // You can make this configurable
AgentID: "1", // You can make this configurable
RegionName: ms.regionName, // Add regional fields
AgentID: ms.agentID,
}
if err := ms.pbClient.SaveDNSData(dnsData); err != nil {
server/service-operation/shared/savers/metrics_saver.go
+58 -3
View File
@@ -1,7 +1,7 @@
package savers
import (
"fmt"
"time"
"service-operation/pocketbase"
@@ -9,12 +9,24 @@ import (
)
type MetricsSaver struct {
pbClient *pocketbase.PocketBaseClient
pbClient *pocketbase.PocketBaseClient
regionName string
agentID string
}
func NewMetricsSaver(pbClient *pocketbase.PocketBaseClient) *MetricsSaver {
return &MetricsSaver{
pbClient: pbClient,
pbClient: pbClient,
regionName: "Default", // Default fallback
agentID: "1", // Default fallback
}
}
func NewMetricsSaverWithRegion(pbClient *pocketbase.PocketBaseClient, regionName, agentID string) *MetricsSaver {
return &MetricsSaver{
pbClient: pbClient,
regionName: regionName,
agentID: agentID,
}
}
@@ -50,6 +62,8 @@ func (ms *MetricsSaver) SaveMetricsToPocketBase(result *types.OperationResult, s
ms.SaveDNSDataToPocketBase(result, serviceID)
case types.OperationTCP:
ms.SaveTCPDataToPocketBase(result, serviceID)
case types.OperationSSL:
ms.SaveSSLDataToPocketBase(result, serviceID)
}
}
}
@@ -86,5 +100,46 @@ func (ms *MetricsSaver) SaveMetricsForService(service pocketbase.Service, result
ms.SaveUptimeDataToPocketBase(result, service.ID)
case "tcp":
ms.SaveTCPDataToPocketBase(result, service.ID)
case "ssl":
ms.SaveSSLDataToPocketBase(result, service.ID)
}
}
// SSL Data saver - remains unchanged (no regional fields)
func (ms *MetricsSaver) SaveSSLDataToPocketBase(result *types.OperationResult, serviceID string) {
// Create SSL data record without regional fields
var details string
if result.Success {
details = fmt.Sprintf("✅ SSL Certificate Valid - Expires in %d days", result.SSLDaysLeft)
details += fmt.Sprintf(" | Valid until: %s", result.SSLValidTill.Format("2006-01-02"))
if result.SSLIssuer != "" {
details += fmt.Sprintf(" | Issuer: %s", result.SSLIssuer)
}
} else {
details = fmt.Sprintf("❌ SSL Certificate Issue - %s", GetShortErrorMessage(result.Error))
}
sslData := pocketbase.SSLDataRecord{
ServiceID: serviceID,
Timestamp: time.Now(),
ResponseTime: result.ResponseTime.Milliseconds(),
Status: GetStatusString(result.Success),
ValidFrom: result.SSLValidFrom.Format(time.RFC3339),
ValidTill: result.SSLValidTill.Format(time.RFC3339),
DaysLeft: result.SSLDaysLeft,
Issuer: result.SSLIssuer,
Subject: result.SSLSubject,
SerialNumber: result.SSLSerialNumber,
Algorithm: result.SSLAlgorithm,
SANs: result.SSLSANs,
ResolvedIP: result.SSLResolvedIP,
ErrorMessage: result.Error,
Details: details,
}
if err := ms.pbClient.SaveSSLData(sslData); err != nil {
println("Failed to save SSL data to PocketBase:", err.Error())
}
}
server/service-operation/shared/savers/ping_saver.go
+2
View File
@@ -59,6 +59,8 @@ func (ms *MetricsSaver) SavePingDataToPocketBase(result *types.OperationResult,
Latency: fmt.Sprintf("%.2fms", float64(result.AvgRTT.Nanoseconds())/1000000),
ErrorMessage: result.Error,
Details: details, // Short, clean message
RegionName: ms.regionName, // Add regional fields
AgentID: ms.agentID,
}
if err := ms.pbClient.SavePingData(pingData); err != nil {
server/service-operation/shared/savers/tcp_saver.go
+2 -2
View File
@@ -51,8 +51,8 @@ func (ms *MetricsSaver) SaveTCPDataToPocketBase(result *types.OperationResult, s
Port: strconv.Itoa(result.Port),
ErrorMessage: result.Error,
Details: details,
RegionName: "default",
AgentID: "1",
RegionName: ms.regionName, // Add regional fields
AgentID: ms.agentID,
}
if err := ms.pbClient.SaveTCPData(tcpData); err != nil {
server/service-operation/shared/savers/uptime_saver.go
+4 -2
View File
@@ -57,8 +57,10 @@ func (ms *MetricsSaver) SaveUptimeDataToPocketBase(result *types.OperationResult
Keyword: "", // Can be populated later if needed
ErrorMessage: result.Error,
Details: details, // Short, clean message
Region: "default", // You can make this configurable
RegionID: "1", // You can make this configurable
Region: ms.regionName, // Legacy field
RegionID: ms.agentID, // Legacy field
RegionName: ms.regionName, // Add regional fields
AgentID: ms.agentID,
}
if err := ms.pbClient.SaveUptimeData(uptimeData); err != nil {