status/service/check.go

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
}