feat: Add check service (WIP)

service/check.go

@@ -0,0 +1,175 @@
+package service
+
+import (
+	"context"
+	"errors"
+	"net"
+	"net/http"
+	"net/url"
+	"slices"
+	"sync"
+	"time"
+
+	"git.ostiwe.com/ostiwe-com/status/model"
+	"git.ostiwe.com/ostiwe-com/status/modules/log"
+	"git.ostiwe.com/ostiwe-com/status/repository"
+	"github.com/sirupsen/logrus"
+)
+
+var (
+	httpObserveFailed   = errors.New("http observe fail")
+	httpObserveMaxTries = errors.New("http observe max tries")
+)
+
+func init() {
+	log.Global.Put(log.CRON, logrus.New())
+}
+
+type Check interface {
+	Start(interval time.Duration)
+}
+
+type check struct {
+	serviceRepository     repository.Service
+	maxServicesPerRoutine int
+	statusRepository      repository.Status
+}
+
+func NewCheck(maxServicesPerRoutine int) Check {
+	if maxServicesPerRoutine <= 0 {
+		maxServicesPerRoutine = 15
+	}
+
+	return &check{
+		maxServicesPerRoutine: maxServicesPerRoutine,
+		serviceRepository:     repository.NewServiceRepository(),
+		statusRepository:      repository.NewStatusRepository(),
+	}
+}
+
+func (c check) Start(interval time.Duration) {
+	for {
+		time.Sleep(interval)
+
+		services, err := c.serviceRepository.All(context.Background(), -1, 0, false)
+		if err != nil {
+			log.Global.Get(log.CRON).Error(err)
+
+			continue
+		}
+
+		wg := new(sync.WaitGroup)
+
+		chunked := slices.Chunk(services, c.maxServicesPerRoutine)
+
+		for i := range chunked {
+			wg.Add(1)
+
+			go c.observeChunk(wg, i)
+		}
+
+		wg.Wait()
+
+	}
+}
+
+func (c check) observeChunk(wg *sync.WaitGroup, chunk []model.Service) {
+	defer wg.Done()
+
+	for _, item := range chunk {
+		switch item.Type {
+		case "http":
+			err := c.observeHttp(item, 0)
+			if err == nil {
+				addErr := c.statusRepository.Add(model.Status{
+					ServiceID: item.ID,
+					Status:    model.StatusOK,
+				})
+				if addErr != nil {
+					log.Global.Get(log.CRON).Error("Add status to status repository fail", item.ID, addErr)
+				}
+
+				continue
+			}
+
+			if errors.Is(err, httpObserveFailed) {
+				addErr := c.statusRepository.Add(model.Status{
+					ServiceID: item.ID,
+					Status:    model.StatusFailed,
+				})
+				if addErr != nil {
+					log.Global.Get(log.CRON).Error("Add status to status repository fail", item.ID, addErr)
+				}
+
+				continue
+			}
+
+			log.Global.Get(log.CRON).Error("Unexpected observe fail", item.ID, err)
+
+			break
+		case "tcp":
+			log.Global.Get(log.CRON).Warn("Cannot handle service with id: ", item.ID, ", because a tcp observer not implemented.")
+			continue
+		}
+	}
+}
+
+func (c check) observeHttp(service model.Service, attempt int) error {
+	if attempt >= 20 {
+		return httpObserveMaxTries
+	}
+
+	if attempt > 5 {
+		err := c.statusRepository.Add(model.Status{
+			ServiceID: service.ID,
+			Status:    model.StatusWarn,
+		})
+		if err != nil {
+			return err
+		}
+	}
+
+	if service.TypeConfig == nil || service.TypeConfig.HTTPConfig == nil {
+		return errors.New("service has no config for http")
+	}
+
+	conf := service.TypeConfig.HTTPConfig
+
+	client := &http.Client{
+		Timeout: time.Second * 5,
+	}
+
+	request, err := http.NewRequest(conf.Method, service.Host, nil)
+	if err != nil {
+		return err
+	}
+
+	for s := range conf.Headers {
+		request.Header.Set(s, conf.Headers[s])
+	}
+
+	response, err := client.Do(request)
+	if err != nil && !errors.Is(err, context.DeadlineExceeded) {
+		var (
+			e     *url.Error
+			opErr *net.OpError
+		)
+		if errors.As(err, &e) && errors.As(e.Err, &opErr) {
+			time.Sleep(2 * time.Second)
+
+			return c.observeHttp(service, attempt+1)
+		}
+
+		return err
+	}
+
+	if errors.Is(err, context.DeadlineExceeded) {
+		return c.observeHttp(service, attempt+1)
+	}
+
+	if response.StatusCode != http.StatusOK {
+		return httpObserveFailed
+	}
+
+	return nil
+}