Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 15 additions & 19 deletions backend/internal/deployment/kubedeployer/client.go
Original file line number Diff line number Diff line change
@@ -1,44 +1,40 @@
package kubedeployer

import (
"fmt"
"kubecloud/internal/infrastructure/gridclient"

"github.com/threefoldtech/tfgrid-sdk-go/grid-client/deployer"
"go.opentelemetry.io/otel/trace"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
)

type Client struct {
GridClient deployer.TFPluginClient
mnemonic string
GridClient gridclient.GridClient
}

func NewClient(mnemonic, gridNet string, debug bool, tp trace.TracerProvider) (*Client, error) {
pluginOpts := []deployer.PluginOpt{
deployer.WithNetwork(gridNet),
deployer.WithDisableSentry(),
// NewClient creates a new Client instance
func NewClient(mnemonic, gridNet string, debug bool, tp *sdktrace.TracerProvider) (*Client, error) {
var opts []gridclient.ClientOpts
if gridNet != "" {
Comment thread
SalmaElsoly marked this conversation as resolved.
Outdated
opts = append(opts, gridclient.WithNetwork(gridNet))
}
if debug {
pluginOpts = append(pluginOpts, deployer.WithLogs())
opts = append(opts, gridclient.WithDebug())
}

if tp != nil {
pluginOpts = append(pluginOpts, deployer.WithTraceProvider(tp))
opts = append(opts, gridclient.WithTracerProvider(tp))
}
opts = append(opts, gridclient.WithDisableSentry())

tfplugin, err := deployer.NewTFPluginClient(
mnemonic,
pluginOpts...,
)
gridCl, err := gridclient.NewGridClient(mnemonic, opts...)
if err != nil {
return nil, fmt.Errorf("failed to create TFPluginClient: %v", err)
return nil, err
}

return &Client{
GridClient: tfplugin,
mnemonic: mnemonic,
GridClient: gridCl,
}, nil
}

// Close closes the underlying GridClient
func (c *Client) Close() {
c.GridClient.Close()
}
22 changes: 11 additions & 11 deletions backend/internal/deployment/kubedeployer/core.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,10 @@ import (
"fmt"
"slices"

"kubecloud/internal/infrastructure/gridclient"
"kubecloud/internal/infrastructure/logger"
"kubecloud/internal/infrastructure/telemetry"

"github.com/threefoldtech/tfgrid-sdk-go/grid-client/deployer"
"github.com/threefoldtech/tfgrid-sdk-go/grid-client/workloads"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
Expand All @@ -24,7 +24,7 @@ func (c *Cluster) GetLeaderNode() (Node, error) {
return c.Nodes[0], nil
}

func (n *Node) AssignNodeIP(ctx context.Context, gridClient deployer.TFPluginClient, networkName string) error {
func (n *Node) AssignNodeIP(ctx context.Context, gridClient gridclient.GridClient, networkName string) error {
ctx, span := getTracer().Start(ctx, "Node.AssignNodeIP",
trace.WithAttributes(
attribute.String("node.name", n.Name),
Expand Down Expand Up @@ -109,8 +109,8 @@ func (c *Client) DeployNode(ctx context.Context, cluster *Cluster, node Node, ma
leaderIP,
cluster.Token,
masterPubKey,
c.mnemonic,
c.GridClient.Network,
c.GridClient.GetMnemonic(),
c.GridClient.GetGridNetwork(),
)
if err != nil {
telemetry.RecordError(span, err)
Expand All @@ -128,7 +128,7 @@ func (c *Client) DeployNode(ctx context.Context, cluster *Cluster, node Node, ma
Str("deployment_name", depl.Name).
Msg("Deploying to grid")

if err := c.GridClient.DeploymentDeployer.Deploy(ctx, &depl); err != nil {
if err := c.GridClient.Deploy(ctx, &depl); err != nil {
log.Error().
Err(err).
Str("node_name", node.Name).
Expand All @@ -144,7 +144,7 @@ func (c *Client) DeployNode(ctx context.Context, cluster *Cluster, node Node, ma
Uint32("node_id", node.NodeID).
Msg("Loading deployment result from grid")

result, err := c.GridClient.State.LoadDeploymentFromGrid(ctx, node.NodeID, node.Name)
result, err := c.GridClient.LoadDeploymentFromGrid(ctx, node.NodeID, node.Name)
if err != nil {
telemetry.RecordError(span, err)
return fmt.Errorf("failed to load deployment for node %s: %v", node.Name, err)
Expand Down Expand Up @@ -241,8 +241,8 @@ func (c *Client) BatchDeployNodes(ctx context.Context, cluster *Cluster, nodes [
leaderIP,
cluster.Token,
masterPubKey,
c.mnemonic,
c.GridClient.Network,
c.GridClient.GetMnemonic(),
c.GridClient.GetGridNetwork(),
)
if err != nil {
telemetry.RecordError(span, err)
Expand All @@ -259,7 +259,7 @@ func (c *Client) BatchDeployNodes(ctx context.Context, cluster *Cluster, nodes [
log.Debug().
Int("deployment_count", len(deployments)).
Msg("Starting batch deployment to grid")
batchErr := c.GridClient.DeploymentDeployer.BatchDeploy(ctx, deployments)
batchErr := c.GridClient.BatchDeploy(ctx, deployments)

var successCount int
var failedNodes []string
Expand All @@ -278,7 +278,7 @@ func (c *Client) BatchDeployNodes(ctx context.Context, cluster *Cluster, nodes [
),
)

result, err := c.GridClient.State.LoadDeploymentFromGrid(ctx, node.NodeID, node.Name)
result, err := c.GridClient.LoadDeploymentFromGrid(ctx, node.NodeID, node.Name)
if err != nil {
log.Warn().Err(err).Str("node_name", node.Name).Msg("Failed to load deployment for node")
failedNodes = append(failedNodes, node.Name)
Expand Down Expand Up @@ -418,7 +418,7 @@ func (c *Client) DeployNetwork(ctx context.Context, cluster *Cluster) error {
Msg("Deploying network")

span.AddEvent("Deploying network")
err = c.GridClient.NetworkDeployer.Deploy(ctx, &net)
err = c.GridClient.DeployNetwork(ctx, &net)
cluster.Network = net
if err != nil {
telemetry.RecordError(span, err)
Expand Down
12 changes: 6 additions & 6 deletions backend/internal/deployment/kubedeployer/netutil.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,15 +5,15 @@ import (
"fmt"
"net"

"kubecloud/internal/infrastructure/gridclient"

"github.com/pkg/errors"
"github.com/threefoldtech/tfgrid-sdk-go/grid-client/deployer"
"github.com/threefoldtech/tfgrid-sdk-go/grid-client/workloads"
"github.com/threefoldtech/tfgrid-sdk-go/grid-client/zos"
)

func getIpForVm(ctx context.Context, tfPluginClient deployer.TFPluginClient, networkName string, nodeID uint32) (string, error) {
network := tfPluginClient.State.Networks.GetNetwork(networkName)
ipRange := network.GetNodeSubnet(nodeID)
func getIpForVm(ctx context.Context, tfPluginClient gridclient.GridClient, networkName string, nodeID uint32) (string, error) {
ipRange := tfPluginClient.GetNodeSubnet(networkName, nodeID)

ip, ipRangeCIDR, err := net.ParseCIDR(ipRange)
if err != nil {
Expand Down Expand Up @@ -46,8 +46,8 @@ func getIpForVm(ctx context.Context, tfPluginClient deployer.TFPluginClient, net
return "", fmt.Errorf("all IPs are exhausted for network %s on node %d", networkName, nodeID)
}

func getUsedHostIDsFromGrid(ctx context.Context, tfPluginClient deployer.TFPluginClient, nodeID uint32, networkName string, ipRangeCIDR *net.IPNet) ([]byte, error) {
nodeClient, err := tfPluginClient.NcPool.GetNodeClient(tfPluginClient.SubstrateConn, nodeID)
func getUsedHostIDsFromGrid(ctx context.Context, tfPluginClient gridclient.GridClient, nodeID uint32, networkName string, ipRangeCIDR *net.IPNet) ([]byte, error) {
nodeClient, err := tfPluginClient.GetNodeClient(nodeID)
if err != nil {
return nil, errors.Wrapf(err, "could not get node client for node %d", nodeID)
}
Expand Down
2 changes: 1 addition & 1 deletion backend/internal/deployment/kubedeployer/noderemove.go
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,7 @@ func (c *Client) isContractActive(ctx context.Context, contractID uint64) bool {

log := logger.ForOperation("kubedeployer", "check_contract_active")
log.Debug().Uint64("contract_id", contractID).Msg("Checking if contract is active")
_, err := c.GridClient.SubstrateConn.GetContract(contractID)
_, err := c.GridClient.GetContract(contractID)
isActive := err == nil

span.SetAttributes(attribute.Bool("contract.is_active", isActive))
Expand Down
9 changes: 8 additions & 1 deletion backend/internal/deployment/statemanager/client_manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (

"github.com/xmonader/ewf"
"go.opentelemetry.io/otel"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
"go.opentelemetry.io/otel/trace/noop"
)

Expand Down Expand Up @@ -65,8 +66,14 @@ func GetKubeClient(state ewf.State, config ClientConfig) (*kubedeployer.Client,
log.Warn().Msg("Tracer provider is no-op, tracing will not work")
}

// Extract SDK trace provider if available
var tp *sdktrace.TracerProvider
if globalTp != nil {
tp, _ = globalTp.(*sdktrace.TracerProvider)
}

// Create new client
kubeClient, err := kubedeployer.NewClient(config.Mnemonic, config.Network, config.Debug, globalTp)
kubeClient, err := kubedeployer.NewClient(config.Mnemonic, config.Network, config.Debug, tp)
if err != nil {
return nil, fmt.Errorf("failed to create kubeclient: %w", err)
}
Expand Down
78 changes: 17 additions & 61 deletions backend/internal/deployment/statemanager/gridclient_state.go
Original file line number Diff line number Diff line change
@@ -1,86 +1,42 @@
package statemanager

import (
"encoding/json"
"fmt"

"kubecloud/internal/deployment/kubedeployer"

"kubecloud/internal/infrastructure/logger"

"github.com/threefoldtech/tfgrid-sdk-go/grid-client/state"
"github.com/xmonader/ewf"
)

// GridClientState represents the critical state that needs to be preserved
type GridClientState struct {
CurrentNodeDeployments map[uint32][]uint64 `json:"current_node_deployments"`
NetworkSubnets map[string]map[uint32]string `json:"network_subnets"`
}
const gridClientStateKey = "gridclient_state"

// SaveGridClientState saves the critical GridClient state to workflow state
// SaveGridClientState saves the GridClient state to the workflow state
func SaveGridClientState(workflowState ewf.State, kubeClient *kubedeployer.Client) {
Comment thread
SalmaElsoly marked this conversation as resolved.
Outdated
log := logger.ForOperation("statemanager", "save_gridclient_state")
if kubeClient == nil || kubeClient.GridClient.State == nil {
if kubeClient == nil || kubeClient.GridClient == nil {
return
}

gridState := GridClientState{
CurrentNodeDeployments: make(map[uint32][]uint64),
NetworkSubnets: make(map[string]map[uint32]string),
}

// Save CurrentNodeDeployments
for nodeID, contractIDs := range kubeClient.GridClient.State.CurrentNodeDeployments {
gridState.CurrentNodeDeployments[nodeID] = []uint64(contractIDs)
}

// Save network subnet information
for networkName, network := range kubeClient.GridClient.State.Networks.State {
gridState.NetworkSubnets[networkName] = network.Subnets
stateData, err := kubeClient.GridClient.GetState()
if err != nil {
return
}

// Store as JSON string in workflow state
if stateBytes, err := json.Marshal(gridState); err == nil {
workflowState["gridclient_state"] = string(stateBytes)
log.Debug().Msg("Saved GridClient state to workflow state")
} else {
log.Warn().Err(err).Msg("Failed to marshal GridClient state")
}
workflowState[gridClientStateKey] = stateData
}

// RestoreGridClientState restores the critical GridClient state from workflow state
// RestoreGridClientState restores the GridClient state from the workflow state
func RestoreGridClientState(workflowState ewf.State, kubeClient *kubedeployer.Client) error {
log := logger.ForOperation("statemanager", "restore_gridclient_state")
if kubeClient == nil || kubeClient.GridClient.State == nil {
return fmt.Errorf("invalid kubeclient or gridclient state")
}

stateStr, ok := workflowState["gridclient_state"].(string)
if !ok || stateStr == "" {
log.Debug().Msg("No GridClient state found in workflow state")
return nil // Not an error, just no state to restore
}

var savedState GridClientState
if err := json.Unmarshal([]byte(stateStr), &savedState); err != nil {
return fmt.Errorf("failed to unmarshal GridClient state: %w", err)
if kubeClient == nil || kubeClient.GridClient == nil {
return nil
}

// Restore CurrentNodeDeployments
kubeClient.GridClient.State.CurrentNodeDeployments = make(map[uint32]state.ContractIDs)
for nodeID, contractIDs := range savedState.CurrentNodeDeployments {
kubeClient.GridClient.State.CurrentNodeDeployments[nodeID] = state.ContractIDs(contractIDs)
stateData, ok := workflowState[gridClientStateKey]
if !ok {
return nil
}

// Restore network subnet information
kubeClient.GridClient.State.Networks.State = make(map[string]state.Network)
for networkName, subnets := range savedState.NetworkSubnets {
kubeClient.GridClient.State.Networks.State[networkName] = state.Network{
Subnets: subnets,
}
stateBytes, ok := stateData.([]byte)
if !ok {
return nil
}

log.Debug().Msg("Restored GridClient state from workflow state")
return nil
return kubeClient.GridClient.RestoreState(stateBytes)
}
Loading
Loading