lobby2/refresher/main.go

package refresher

import (
	"bytes"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"path"
	"time"

	"gitea.ceperka.net/rosti/lobby2/nodes"
)

const refreshIntervalSeconds = 15
const nodeFileName = "node.json"

// Refresher loads local node info and sends them to the master node.
type Refresher struct {
	configPath  string
	nodeDirPath string
}

func NewRefresher(nodeDirPath string, configPath string) *Refresher {
	return &Refresher{
		nodeDirPath: nodeDirPath,
		configPath:  configPath,
	}
}

// Run loop the process that updates node in the master node.
func (r *Refresher) Loop() {
	log.Println("Start refresher loop")

	for {
		log.Println("Refreshing node")
		err := r.Refresh()
		if err != nil {
			log.Printf("failed to refresh node: %v\n", err)
		}
		time.Sleep(time.Second * refreshIntervalSeconds)
	}
}

// Refresh loads labels from the local filesystem and sends them to the master node.
func (r *Refresher) Refresh() error {
	// Load config
	cfg, err := r.getConfig()
	if err != nil {
		return err
	}

	// Load labels
	node, err := r.loadNode()
	if err != nil {
		return err
	}

	nodeBytes, err := json.Marshal(node)
	if err != nil {
		return fmt.Errorf("failed to marshal labels: %w", err)
	}

	// Send labels to master
	req, err := http.NewRequest("POST", fmt.Sprintf("%s://%s:%d/nodes/%s", cfg.MasterProto, cfg.MasterHost, cfg.MasterPort, node.HostName), bytes.NewBuffer(nodeBytes))
	if err != nil {
		return err
	}

	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+cfg.MasterToken)

	client := &http.Client{}
	resp, err := client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusNoContent {
		return fmt.Errorf("unexpected status: %s", resp.Status)
	}

	return nil

}

// Returns the node config.
func (r *Refresher) getConfig() (NodeConfig, error) {
	cfg := NodeConfig{}

	content, err := os.ReadFile(r.configPath)
	if err != nil {
		return cfg, fmt.Errorf("failed to read config file: %w", err)
	}

	err = json.Unmarshal(content, &cfg)
	if err != nil {
		return cfg, fmt.Errorf("failed to unmarshal config: %w", err)
	}

	if cfg.MasterProto == "" {
		cfg.MasterProto = "http"
	}

	if cfg.MasterPort == 0 {
		cfg.MasterPort = 1352
	}

	return cfg, nil
}

// TODO: rewrite this to load Node structure
func (r *Refresher) loadNode() (*nodes.Node, error) {
	filePath := path.Join(r.nodeDirPath, nodeFileName)
	_, err := os.Stat(filePath)
	if os.IsNotExist(err) {
		err = r.initNodeFile()
		if err != nil {
			return nil, err
		}
	}

	content, err := os.ReadFile(path.Join(r.nodeDirPath, nodeFileName))
	if err != nil {
		return nil, fmt.Errorf("failed to read file: %w", err)
	}

	node := &nodes.Node{}
	err = json.Unmarshal(content, &node)
	if err != nil {
		return nil, fmt.Errorf("failed to unmarshal node: %w", err)
	}

	return node, err
}

func (r *Refresher) initNodeFile() error {
	err := r.createNodePath()
	if err != nil {
		return fmt.Errorf("failed to create node path: %w", err)
	}

	hostname, err := os.Hostname()
	if err != nil {
		return fmt.Errorf("failed to get hostname: %w", err)
	}

	node := &nodes.Node{
		HostName: hostname,
	}

	nodeBytes, err := json.MarshalIndent(node, "", "  ")
	if err != nil {
		return fmt.Errorf("failed to marshal node: %w", err)
	}

	err = os.WriteFile(path.Join(r.nodeDirPath, nodeFileName), nodeBytes, 0640)
	if err != nil {
		return fmt.Errorf("failed to write node file: %w", err)
	}

	return nil
}

func (r *Refresher) createNodePath() error {
	_, err := os.Stat(r.nodeDirPath)
	if os.IsNotExist(err) {
		err = os.MkdirAll(r.nodeDirPath, 0755)
		if err != nil {
			return fmt.Errorf("failed to create node dir: %w", err)
		}
	}

	return nil

}