Merge branch 'master' into chore/retry-processor

backend/alerter.go

@@ -0,0 +1,87 @@
+package main
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "time"
+)
+
+type Alerter interface {
+ Send(ctx context.Context, msg AlertMessage) error
+}
+
+type AlertMessage struct {
+ Success bool
+ StatusCode int
+ Timestamp time.Time
+ MonitorID string
+ MonitorName string
+ Latency int64
+}
+
+type TelegramProvider struct {
+ url string
+ chatID string
+}
+
+type TelegramProviderConfig struct {
+ Url string
+ ChatID string
+}
+
+func NewTelegramAlertProvider(config TelegramProviderConfig) *TelegramProvider {
+ return &TelegramProvider{
+ url: config.Url,
+ }
+}
+
+func (t TelegramProvider) Send(ctx context.Context, msg AlertMessage) error {
+ if t.url == "" || t.chatID == "" {
+ return fmt.Errorf("can't make a telegram alert request: some config is not set")
+ }
+
+ // Perhaps we can use a template file instead.
+ title := "🔴 Down"
+ if msg.Success {
+ title = "✅ Up"
+ }
+ text := fmt.Sprintf(title+`
+
+ **MonitorID:** %s
+ **MonitorName:** %s
+ **StatusCode:** %d
+ **Latency:** %d
+ **Timestamp:** %s`,
+ msg.MonitorID,
+ msg.MonitorName,
+ msg.StatusCode,
+ msg.Latency,
+ msg.Timestamp.Format(time.RFC3339),
+ )
+ payload := map[string]any{
+ "chat_id": t.chatID,
+ "text": text,
+ "parse_mode": "Markdown",
+ }
+ payloadByte, _ := json.Marshal(payload)
+
+ req, err := http.NewRequestWithContext(ctx, http.MethodPost, t.url, bytes.NewReader(payloadByte))
+ if err != nil {
+ return fmt.Errorf("failed to send telegram alert: %w", err)
+ }
+ defer req.Body.Close()
+
+ req.Header.Set("Content-Type", "application/json")
+
+ client := http.Client{Timeout: time.Second * 3}
+
+ _, err = client.Do(req)
+ if err != nil {
+ return fmt.Errorf("failed to make request: %w", err)
+ }
+
+ return nil
+}

backend/config.go

@@ -25,6 +25,14 @@ const (
  MonitorTypePing MonitorType = "ping"
 )
 
+type AlertProviderType string
+
+const (
+ AlertProviderTypeUnspecified AlertProviderType = ""
+ AlertProviderTypeTelegram AlertProviderType = "telegram"
+ AlertProviderTypeDiscord AlertProviderType = "discord"
+)
+
 type Monitor struct {
  // UniqueID specifies unique identifier for each monitor. In any case of the monitor configuration value get
  // changed (name, description, public monitorIds, etc), if users want to keep the data intact, they should keep the
@@ -61,6 +69,10 @@ type Monitor struct {
  // IcmpPacketSize specifies the packet size that will be used for the ICMP request. It must be greater than zero.
  // The default packet size is 56 bytes.
  IcmpPacketSize int `json:"packet_size" yaml:"packet_size" toml:"packet_size"`
+ // AlertProvider specifies the type of alert provider that will be used to send alerts. It can be a string value such as
+ // "telegram" or "discord".
+ // THe default alert provider is "telegram"
+ AlertProvider AlertProviderType `json:"alert_provider" yam:"alert_provider" toml:"alert_provider"`
 }
 
 func (m Monitor) MarshalJSON() ([]byte, error) {

backend/incident_reader.go

@@ -0,0 +1,77 @@
+package main
+
+import (
+ "context"
+ "database/sql"
+ "fmt"
+ "time"
+
+ "github.com/rs/zerolog/log"
+)
+
+type IncidentDataReader struct {
+ db *sql.DB
+}
+
+func NewIncidentDataReader(db *sql.DB) *IncidentDataReader {
+ return &IncidentDataReader{db: db}
+}
+
+func (r *IncidentDataReader) ReadRelatedIncidents(ctx context.Context, incidentTitle string, monitorID string) ([]Incident, error) {
+ dbCon, err := r.db.Conn(ctx)
+ if err != nil {
+ return nil, err
+ }
+ defer func() {
+ err := dbCon.Close()
+ if err != nil {
+ log.Warn().Stack().Err(err).Msg("Failed to close connection")
+ }
+ }()
+
+ rows, err := dbCon.QueryContext(ctx, "SELECT monitor_id, title, description, timestamp, severity, status FROM incident_data WHERE monitor_id = ? AND title = ? ORDER BY created_at DESC", monitorID, incidentTitle)
+ if err != nil {
+ return nil, fmt.Errorf("failed to read related incidents: %w", err)
+ }
+ defer func() {
+ err := rows.Close()
+ if err != nil {
+ log.Warn().Stack().Err(err).Msg("Failed to close rows")
+ }
+ }()
+
+ var incidents []Incident
+ for rows.Next() {
+ var incident Incident
+ err := rows.Scan(&incident.MonitorID, &incident.Title, &incident.Description, &incident.Timestamp, &incident.Severity, &incident.Status)
+ if err != nil {
+ return nil, err
+ }
+ incidents = append(incidents, incident)
+ }
+
+ return incidents, nil
+}
+
+func (r *IncidentDataReader) ReadIncidentByTimestamp(ctx context.Context, incidentTitle string, monitorID string, timestamp time.Time) (Incident, error) {
+ dbCon, err := r.db.Conn(ctx)
+ if err != nil {
+ return Incident{}, err
+ }
+ defer func() {
+ err := dbCon.Close()
+ if err != nil {
+ log.Warn().Stack().Err(err).Msg("Failed to close connection")
+ }
+ }()
+
+ var incidentDetail Incident
+ err = dbCon.
+ QueryRowContext(ctx, "SELECT monitor_id, title, description, timestamp, severity, status FROM incident_data WHERE monitor_id = ? AND title = ? AND timestamp = ?", monitorID, incidentTitle, timestamp).
+ Scan(incidentDetail.MonitorID, incidentDetail.Title, incidentDetail.Description, incidentDetail.Timestamp, incidentDetail.Severity, incidentDetail.Status)
+ if err != nil {
+ return Incident{}, err
+ }
+
+ return incidentDetail, nil
+}

backend/main.go

@@ -58,6 +58,16 @@ func main() {
  log.Warn().Msg("API_KEY is not set")
  }
 
+ telegramChatID, ok := os.LookupEnv("TELEGRAM_CHAT_ID")
+ if !ok {
+ log.Warn().Msg("TELEGRAM_CHAT_ID is not set")
+ }
+
+ telegramUrl, ok := os.LookupEnv("TELEGRAM_URL is not set")
+ if !ok {
+ log.Warn().Msg("TELEGRAM_URL is not set")
+ }
+
  if os.Getenv("ENV") == "" {
  err := os.Setenv("ENV", "development")
  if err != nil {
@@ -100,7 +110,12 @@ func main() {
  log.Fatal().Err(err).Msg("failed to migrate database")
  }
 
- processor := &Processor{}
+ processor := &Processor{
+ telegramAlertProvider: NewTelegramAlertProvider(TelegramProviderConfig{
+ Url: telegramUrl,
+ ChatID: telegramChatID,
+ }),
+ }
 
  // Create a new worker
  for _, monitor := range config.Monitors {

backend/monitor_processor.go

@@ -11,6 +11,9 @@ import (
 type Processor struct {
  historicalWriter *MonitorHistoricalWriter
  historicalReader *MonitorHistoricalReader
+
+ telegramAlertProvider Alerter
+ discordAlertProvider Alerter
 }
 
 func (m *Processor) ProcessResponse(response Response) {
@@ -53,5 +56,42 @@ func (m *Processor) ProcessResponse(response Response) {
  break
  }
 
- // TODO: If the current status is different from the last status, send an alert notification
-}
+ go func() {
+ if m.telegramAlertProvider == nil && m.discordAlertProvider == nil {
+ log.Warn().Msg("no alert providers are set")
+ return
+ }
+
+ alertMessage := AlertMessage{
+ Success: response.Success,
+ MonitorID: uniqueId,
+ MonitorName: response.Monitor.Name,
+ StatusCode: response.StatusCode,
+ Timestamp: response.Timestamp,
+ Latency: response.RequestDuration,
+ }
+
+ lastRawHistorical, err := m.historicalReader.ReadRawLatest(context.Background(), uniqueId)
+ if err != nil {
+ log.Error().Err(err).Msg("failed to get raw latest historical data")
+ return
+ }
+
+ if lastRawHistorical.Status != status {
+ switch response.Monitor.AlertProvider {
+ case AlertProviderTypeTelegram, AlertProviderTypeUnspecified:
+ if m.telegramAlertProvider == nil {
+ log.Warn().Msg("telegram alert provider is not set")
+ return
+ }
+
+ err := m.telegramAlertProvider.Send(context.Background(), alertMessage)
+ if err != nil {
+ log.Error().Err(err).Msg("failed to send alert")
+ }
+ case AlertProviderTypeDiscord:
+ panic("TODO: Implement me!")
+ }
+ }
+ }()
+}