compute/pkg/raft/store.go

package raft

import (
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"sync"
	"time"

	transport "deevirt.fr/compute/pkg/raft/transport"
	"github.com/hashicorp/raft"
	raftboltdb "github.com/hashicorp/raft-boltdb/v2"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"

	"deevirt.fr/compute/pkg/config"
	etcd_client "deevirt.fr/compute/pkg/etcd"
	"deevirt.fr/compute/pkg/scheduler"
)

type RaftNode struct {
	Raft      *raft.Raft
	NodeID    string
	StateCh   chan raft.Observation
	scheduler *scheduler.Scheduler
}

type Peers struct {
	Id      string
	Address string
}

func getTLSCredentials(conf *config.Config) credentials.TransportCredentials {
	cert, err := tls.LoadX509KeyPair(conf.Manager.TlsCert, conf.Manager.TlsKey)
	if err != nil {
		log.Fatalf("Erreur chargement du certificat: %v", err)
	}

	// Charger la CA (facultatif, pour la vérification des clients)
	caCert, err := os.ReadFile(conf.Manager.TlsCert)
	if err != nil {
		log.Fatalf("Erreur chargement CA: %v", err)
	}
	certPool := x509.NewCertPool()
	certPool.AppendCertsFromPEM(caCert)

	// Créer les credentials TLS
	creds := credentials.NewTLS(&tls.Config{
		Certificates:       []tls.Certificate{cert},
		ClientCAs:          certPool,
		InsecureSkipVerify: true,
	})

	return creds
}

const (
	retainSnapshotCount = 2
	raftTimeout         = 10 * time.Second
)

type command struct {
	Op    string `json:"op,omitempty"`
	Key   string `json:"key,omitempty"`
	Value string `json:"value,omitempty"`
}

type Store struct {
	mu   sync.Mutex
	conf *config.Config // Configuration générale
	port int            // Port de communication (identique au serveur GRPC)

	m map[string]string // The key-value store for the system.

	Raft *raft.Raft // The consensus mechanism

	logger *log.Logger
}

func New(conf *config.Config, port int) *Store {
	return &Store{
		conf:   conf,
		port:   port,
		m:      make(map[string]string),
		logger: log.New(os.Stderr, "[store] ", log.LstdFlags),
	}
}

func (s *Store) Open() (*transport.Manager, error) {
	// Création du répertoire
	baseDir := filepath.Join("/var/lib/deevirt/mgr/", s.conf.NodeID)
	err := os.MkdirAll(baseDir, 0740)
	if err != nil {
		return nil, err
	}

	// Récupération des Noeuds ID
	etcd, err := etcd_client.New(s.conf.EtcdURI)
	if err != nil {
		return nil, err
	}
	defer etcd.Close()

	peers := []raft.Server{}

	for key, value := range etcd_client.GetNodes(etcd, s.conf.ClusterID) {
		var p string

		for _, peer := range s.conf.Manager.Peers {
			if peer == value.IpManagement {
				p = peer
			}
		}

		if p != "" {
			peers = append(peers, raft.Server{
				ID:      raft.ServerID(key),
				Address: raft.ServerAddress(fmt.Sprintf("%s:%d", p, s.port)),
			})
		}
	}

	c := raft.DefaultConfig()
	c.LocalID = raft.ServerID(s.conf.NodeID)

	ldb, err := raftboltdb.NewBoltStore(filepath.Join(baseDir, "logs.dat"))
	if err != nil {
		return nil, fmt.Errorf(`boltdb.NewBoltStore(%q): %v`, filepath.Join(baseDir, "logs.dat"), err)
	}

	fss, err := raft.NewFileSnapshotStore(baseDir, 3, os.Stderr)
	if err != nil {
		return nil, fmt.Errorf(`raft.NewFileSnapshotStore(%q, ...): %v`, baseDir, err)
	}

	dialOption := []grpc.DialOption{}

	if s.conf.Manager.TlsKey != "" {
		dialOption = append(dialOption, grpc.WithTransportCredentials(getTLSCredentials(s.conf)))
	}

	tm := transport.New(raft.ServerAddress(fmt.Sprintf("%s:%d", s.conf.AddressPrivate, s.port)), dialOption)

	r, err := raft.NewRaft(c, (*Fsm)(s), ldb, ldb, fss, tm.Transport())
	if err != nil {
		return nil, fmt.Errorf("raft.NewRaft: %v", err)
	}
	s.Raft = r

	sched, err := scheduler.New()
	if err != nil {
		return nil, fmt.Errorf("scheduler: %v", err)
	}

	// Observer pour surveiller les changements d'état
	stateCh := make(chan raft.Observation, 1) // Canal de type raft.Observation
	r.RegisterObserver(raft.NewObserver(stateCh, true, nil))

	node := &RaftNode{
		Raft:      r,
		NodeID:    s.conf.NodeID,
		StateCh:   stateCh,
		scheduler: sched,
	}

	go node.watchStateChanges()

	hasState, _ := checkIfStateExists(ldb)

	if s.conf.Manager.Peers[0] == s.conf.AddressPrivate && !hasState {
		println("Démarrage du bootstrap ! ")

		cfg := raft.Configuration{
			Servers: peers,
		}
		f := r.BootstrapCluster(cfg)
		if err := f.Error(); err != nil {
			return nil, fmt.Errorf("raft.Raft.BootstrapCluster: %v", err)
		}
	}

	return tm, nil
}

// Get returns the value for the given key.
func (s *Store) Get(key string) (string, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.m[key], nil
}

// Set sets the value for the given key.
func (s *Store) Set(key, value string) error {
	if s.Raft.State() != raft.Leader {
		return fmt.Errorf("not leader")
	}

	c := &command{
		Op:    "set",
		Key:   key,
		Value: value,
	}
	b, err := json.Marshal(c)
	if err != nil {
		return err
	}

	f := s.Raft.Apply(b, raftTimeout)
	return f.Error()
}

// Delete deletes the given key.
func (s *Store) Delete(key string) error {
	if s.Raft.State() != raft.Leader {
		return fmt.Errorf("not leader")
	}

	c := &command{
		Op:  "delete",
		Key: key,
	}
	b, err := json.Marshal(c)
	if err != nil {
		return err
	}

	f := s.Raft.Apply(b, raftTimeout)
	return f.Error()
}

// Vérifie si l'état Raft existe déjà
func checkIfStateExists(logStore *raftboltdb.BoltStore) (bool, error) {
	// Vérifier les logs Raft
	firstIndex, err := logStore.FirstIndex()
	if err != nil {
		return false, err
	}

	if firstIndex > 0 {
		return true, nil
	}

	return false, nil
}

// Fonction pour surveiller et afficher les changements d'état
func (n *RaftNode) watchStateChanges() {
	for obs := range n.StateCh {
		switch evt := obs.Data.(type) {
		case raft.RaftState:

			if evt == raft.Leader {
				go n.scheduler.Start()

				log.Println("[ÉVÉNEMENT] Changement d'état Raft :", evt)
			} else {
				n.scheduler.Stop()
			}

			log.Println("[ÉVÉNEMENT] Changement d'état Raft :", evt)
		case raft.LeaderObservation:
			log.Println("[ÉVÉNEMENT] Le leader est", evt.LeaderID)
		case raft.PeerObservation:
			if n.Raft.State() == raft.Leader {
				peerID := evt.Peer.ID
				peerAddr := evt.Peer.Address

				log.Println("[NOUVEAU NŒUD] Détection de", peerID, "à", peerAddr)
				log.Println("[ACTION] Ajout automatique en tant que voter...")

				future := n.Raft.AddVoter(peerID, peerAddr, 0, 0)
				if err := future.Error(); err != nil {
					log.Println("[ERREUR] Impossible d'ajouter", peerID, ":", err)
				} else {
					log.Println("[SUCCÈS] Voter ajouté :", peerID)
				}
			}
		case raft.FailedHeartbeatObservation:
			log.Println("[ÉVÉNEMENT] Perte de connexion avec un nœud :", evt.PeerID)
		default:
			log.Println("[ÉVÉNEMENT] Autre événement :", evt)
		}
	}
}

/*func New(ctx context.Context, myID, myAddress string) (*raft.Raft, *transport.Manager, error) {
	// Création du répertoire
	baseDir := filepath.Join("/var/lib/deevirt/mgr/", myID)
	err := os.MkdirAll(baseDir, 0740)
	if err != nil {
		return nil, nil, err
	}

	println(myAddress)

	peers := []raft.Server{
		{
			ID:      raft.ServerID("nodeA"),
			Address: raft.ServerAddress("172.16.9.161:4410"),
		},
		{
			ID:      raft.ServerID("nodeB"),
			Address: raft.ServerAddress("172.16.9.161:4411"),
		},
	}

	c := raft.DefaultConfig()
	c.LocalID = raft.ServerID(myID)

	ldb, err := raftboltdb.NewBoltStore(filepath.Join(baseDir, "logs.dat"))
	if err != nil {
		return nil, nil, fmt.Errorf(`boltdb.NewBoltStore(%q): %v`, filepath.Join(baseDir, "logs.dat"), err)
	}

	sdb, err := raftboltdb.NewBoltStore(filepath.Join(baseDir, "stable.dat"))
	if err != nil {
		return nil, nil, fmt.Errorf(`boltdb.NewBoltStore(%q): %v`, filepath.Join(baseDir, "stable.dat"), err)
	}

	fss, err := raft.NewFileSnapshotStore(baseDir, 3, os.Stderr)
	if err != nil {
		return nil, nil, fmt.Errorf(`raft.NewFileSnapshotStore(%q, ...): %v`, baseDir, err)
	}

	tm := transport.New(raft.ServerAddress(myAddress), []grpc.DialOption{grpc.WithTransportCredentials(getTLSCredentials())})

	r, err := raft.NewRaft(c, nil, ldb, sdb, fss, tm.Transport())
	if err != nil {
		return nil, nil, fmt.Errorf("raft.NewRaft: %v", err)
	}

	s, err := scheduler.New()
	if err != nil {
		return nil, nil, fmt.Errorf("scheduler: %v", err)
	}

	// Observer pour surveiller les changements d'état
	stateCh := make(chan raft.Observation, 1) // Canal de type raft.Observation
	r.RegisterObserver(raft.NewObserver(stateCh, true, nil))

	node := &RaftNode{
		Raft:      r,
		NodeID:    myID,
		StateCh:   stateCh,
		scheduler: s,
	}

	go node.watchStateChanges()

	hasState, _ := checkIfStateExists(ldb)

	if myAddress == "172.16.9.161:4410" && !hasState {
		println("Démarrage du bootstrap ! ")

		cfg := raft.Configuration{
			Servers: peers,
		}
		f := r.BootstrapCluster(cfg)
		if err := f.Error(); err != nil {
			return nil, nil, fmt.Errorf("raft.Raft.BootstrapCluster: %v", err)
		}
	}

	return r, tm, nil
}*/