compute/pkg/api/domain/domain.go

package domain

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"regexp"

	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"libvirt.org/go/libvirt"

	"deevirt.fr/compute/pkg/api/proto"
	"deevirt.fr/compute/pkg/api/raft"
	"deevirt.fr/compute/pkg/config"
	"deevirt.fr/compute/pkg/scheduler"
)

type Domain struct {
	Config *config.Config
	Store  *raft.Store
	proto.UnimplementedDomainServer
}

func (d *Domain) connectNode(NodeId string) (*libvirt.Connect, error) {
	var jCluster raft.NodeStore
	cluster, _ := d.Store.Get("/etc/libvirt/cluster")
	json.Unmarshal(cluster, &jCluster)

	var libvirt_uri string
	if d.Config.LibvirtTLS {
		libvirt_uri = fmt.Sprintf("qemu+tls://%s/system", jCluster[NodeId].IpManagement)
	} else {
		libvirt_uri = fmt.Sprintf("qemu+tcp://%s/system", jCluster[NodeId].IpManagement)
	}

	c, err := libvirt.NewConnect(libvirt_uri)
	if err != nil {
		log.Fatalf("Erreur %v", err)
	}

	return c, nil
}

func (d *Domain) connectDomain(ctx context.Context, domainID string) (*libvirt.Connect, error) {
	dom, _ := d.Get(ctx, &proto.DomainListRequest{
		DomainId: domainID,
	})

	var jCluster raft.NodeStore
	cluster, _ := d.Store.Get("/etc/libvirt/cluster")
	json.Unmarshal(cluster, &jCluster)

	return d.connectNode(dom.NodeId)
}

func (d *Domain) List(ctx context.Context, in *proto.DomainListAllRequest) (*proto.DomainListAllResponse, error) {
	test, _ := d.Store.Ls("/etc/libvirt/qemu", raft.LsOptions{
		Recursive: true,
		Data:      true,
	})

	domains := []*proto.DomainListResponse{}

	for k, v := range test {
		re := regexp.MustCompile("([^/]+)/([^/]+)")
		matches := re.FindStringSubmatch(k)
		if matches != nil {
			var dStore raft.DomainStore
			json.Unmarshal(v, &dStore)

			domains = append(domains, &proto.DomainListResponse{
				NodeId:   matches[1],
				DomainId: matches[2],
				Config:   dStore.Config,
				State:    int64(dStore.State),
			})
		}
	}

	return &proto.DomainListAllResponse{
		Domains: domains,
	}, nil
}

func (d *Domain) Get(ctx context.Context, in *proto.DomainListRequest) (*proto.DomainListResponse, error) {
	dom, _ := d.Store.Ls("/etc/libvirt/qemu", raft.LsOptions{
		Recursive: true,
		Data:      false,
	})

	for k := range dom {
		re := regexp.MustCompile("([^/]+)/([^/]+)")
		matches := re.FindStringSubmatch(k)
		if matches != nil && matches[2] == in.DomainId {
			var dStore raft.DomainStore
			data, _ := d.Store.Get(fmt.Sprintf("/etc/libvirt/qemu/%s/%s", matches[1], matches[2]))
			json.Unmarshal(data, &dStore)

			return &proto.DomainListResponse{
				NodeId:   matches[1],
				DomainId: matches[2],
				Config:   dStore.Config,
				State:    int64(dStore.State),
			}, nil
		}
	}

	return &proto.DomainListResponse{}, nil
}

func (d *Domain) Migrate(in *proto.DomainMigrateRequest, stream proto.Domain_MigrateServer) error {
	libvirt.EventRegisterDefaultImpl()
	ctx := context.Background()

	c, err := d.connectDomain(ctx, in.DomainId)
	if err != nil {
		return status.Errorf(codes.Internal, "Connexion error to libvirt")
	}
	defer c.Close()

	dom, err := c.LookupDomainByUUIDString(in.DomainId)
	if err != nil {
		return status.Errorf(codes.Internal, "Domain unknown")
	}

	s, err := scheduler.New()
	if err != nil {
		return status.Errorf(codes.Internal, "Connexion error to libvirt %v", err)
	}
	res, err := s.GetTopNode(1)
	if err != nil {
		return status.Errorf(codes.Internal, "Connexion error to libvirt %v", err)
	}

	eventCallback := func(cc *libvirt.Connect, dd *libvirt.Domain, ee *libvirt.DomainEventMigrationIteration) {
		// Créer et envoyer le message de progression de la migration
		stream.Send(&proto.DomainMigrateResponse{
			Percentage: int32(ee.Iteration),
		})

		t, _ := dd.QemuMonitorCommand("{\"execute\": \"query-migrate\"}", libvirt.DOMAIN_QEMU_MONITOR_COMMAND_DEFAULT)
		fmt.Printf("%v\n", t)
	}

	// Enregistrer l'événement de migration
	c.DomainEventMigrationIterationRegister(dom, eventCallback)

	ctx1, cancel := context.WithCancel(context.Background())

	migrate := func() error {
		c_new, err := d.connectNode(res[0].NodeID)
		if err != nil {
			cancel()
			return status.Errorf(codes.Internal, "Connexion error to libvirt %v", err)
		}
		defer c_new.Close()

		_, err = dom.Migrate(c_new, libvirt.MIGRATE_LIVE|libvirt.MIGRATE_PERSIST_DEST|libvirt.MIGRATE_UNDEFINE_SOURCE, "", "", 0)
		if err != nil {
			cancel()
			return status.Errorf(codes.Internal, "Migration error %v", err)
		}

		cancel()

		return nil
	}

	go migrate()

	for {
		select {
		case <-ctx1.Done():
			return nil
		default:
			libvirt.EventRunDefaultImpl()
		}
	}
}