Skip to content

Commit

Permalink
deploy max-workers (#176)
Browse files Browse the repository at this point in the history 
* add max-workers flag to deploy cmd

* fix debug statement format

* limit max workers to number of nodes or number of links

* added max-workders doc (#181)

Co-authored-by: Roman Dodin <dodin.roman@gmail.com>
  • Loading branch information
@karimra @hellt
karimra and hellt authored Nov 28, 2020
1 parent e8b5dd3 commit eaa45e1
Show file tree
Hide file tree
Showing 2 changed files with 97 additions and 32 deletions.
124 changes: 93 additions & 31 deletions cmd/deploy.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,9 @@ var mgmtIPv6Subnet net.IPNet
// reconfigure flag
var reconfigure bool

// max-workers flag
var maxWorkers uint

// deployCmd represents the deploy command
var deployCmd = &cobra.Command{
Use: "deploy",
Expand Down Expand Up @@ -89,60 +92,118 @@ var deployCmd = &cobra.Command{
if err != nil {
return fmt.Errorf("failed to parse certCsrTemplate: %v", err)
}
// create directory structure and container per node

numNodes := uint(len(c.Nodes))
numLinks := uint(len(c.Links))
nodesMaxWorkers := maxWorkers
linksMaxWorkers := maxWorkers

if maxWorkers == 0 {
nodesMaxWorkers = numNodes
linksMaxWorkers = numLinks
}

if nodesMaxWorkers > numNodes {
nodesMaxWorkers = numNodes
}
if linksMaxWorkers > numLinks {
linksMaxWorkers = numLinks
}

wg := new(sync.WaitGroup)
wg.Add(len(c.Nodes))
for _, node := range c.Nodes {
go func(node *clab.Node) {
wg.Add(int(nodesMaxWorkers))
nodesChan := make(chan *clab.Node)
// start workers
for i := uint(0); i < nodesMaxWorkers; i++ {
go func(i uint) {
defer wg.Done()
if node.Kind == "bridge" {
return
}
// create CERT
nodeCerts, err := c.GenerateCert(
path.Join(c.Dir.LabCARoot, "root-ca.pem"),
path.Join(c.Dir.LabCARoot, "root-ca-key.pem"),
certTpl,
node,
)
if err != nil {
log.Errorf("failed to generate certificates for node %s: %v", node.ShortName, err)
for {
select {
case node := <-nodesChan:
if node == nil {
log.Debugf("Worker %d terminating...", i)
return
}
log.Debugf("Worker %d received node: %+v", i, node)
if node.Kind == "bridge" {
return
}
// create CERT
nodeCerts, err := c.GenerateCert(
path.Join(c.Dir.LabCARoot, "root-ca.pem"),
path.Join(c.Dir.LabCARoot, "root-ca-key.pem"),
certTpl,
node,
)
if err != nil {
log.Errorf("failed to generate certificates for node %s: %v", node.ShortName, err)
}
log.Debugf("%s CSR: %s", node.ShortName, string(nodeCerts.Csr))
log.Debugf("%s Cert: %s", node.ShortName, string(nodeCerts.Cert))
log.Debugf("%s Key: %s", node.ShortName, string(nodeCerts.Key))
err = c.CreateNode(ctx, node, nodeCerts)
if err != nil {
log.Errorf("failed to create node %s: %v", node.ShortName, err)
}
case <-ctx.Done():
return
}
}
log.Debugf("%s CSR: %s", node.ShortName, string(nodeCerts.Csr))
log.Debugf("%s Cert: %s", node.ShortName, string(nodeCerts.Cert))
log.Debugf("%s Key: %s", node.ShortName, string(nodeCerts.Key))
err = c.CreateNode(ctx, node, nodeCerts)
if err != nil {
log.Errorf("failed to create node %s: %v", node.ShortName, err)
}
}(node)
}(i)
}
for _, n := range c.Nodes {
nodesChan <- n
}
// close channel to terminate the workers
close(nodesChan)
// wait for all workers to finish
wg.Wait()

// cleanup hanging resources if a deployment failed before
log.Debug("cleaning up interfaces...")
c.InitVirtualWiring()
wg = new(sync.WaitGroup)
wg.Add(len(c.Links))
wg.Add(int(linksMaxWorkers))
linksChan := make(chan *clab.Link)
log.Debug("creating links...")
// wire the links between the nodes based on cabling plan
for _, link := range c.Links {
go func(link *clab.Link) {
for i := uint(0); i < linksMaxWorkers; i++ {
go func(i uint) {
defer wg.Done()
if err = c.CreateVirtualWiring(link); err != nil {
log.Error(err)
for {
select {
case link := <-linksChan:
if link == nil {
log.Debugf("Worker %d terminating...", i)
return
}
log.Debugf("Worker %d received link: %+v", i, link)
if err = c.CreateVirtualWiring(link); err != nil {
log.Error(err)
}
case <-ctx.Done():
return
}
}
}(link)
}(i)
}
for _, link := range c.Links {
linksChan <- link
}
// close channel to terminate the workers
close(linksChan)
// wait for all workers to finish
wg.Wait()

// generate graph of the lab topology
if graph {
if err = c.GenerateGraph(topo); err != nil {
log.Error(err)
}
}
log.Debug("containers created, retrieving state and IP addresses...")
// show topology output

// show topology output
labels = append(labels, "containerlab=lab-"+c.Config.Name)
containers, err := c.ListContainers(ctx, labels)
if err != nil {
Expand All @@ -169,6 +230,7 @@ func init() {
deployCmd.Flags().IPNetVarP(&mgmtIPv4Subnet, "ipv4-subnet", "4", net.IPNet{}, "management network IPv4 subnet range")
deployCmd.Flags().IPNetVarP(&mgmtIPv6Subnet, "ipv6-subnet", "6", net.IPNet{}, "management network IPv6 subnet range")
deployCmd.Flags().BoolVarP(&reconfigure, "reconfigure", "", false, "regenerate configuration artifacts and overwrite the previous ones if any")
deployCmd.Flags().UintVarP(&maxWorkers, "max-workers", "", 0, "limit the maximum number of workers creating nodes and virtual wires")
}

func setFlags(conf *clab.Config) {
Expand Down
5 changes: 4 additions & 1 deletion docs/cmd/deploy.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,9 @@ Without this flag present, containerlab will reuse the available configuration a

Refer to the [configuration artifacts](../manual/conf-artifacts.md) page to get more information on the lab directory contents.

#### max-workers
With `--max-workers` flag it is possible to limit the amout of concurrent workers that create containers or wire virtual links. By default the number of workers equals the number of nodes/links to create.

### Examples

```bash
Expand All @@ -36,4 +39,4 @@ containerlab deploy -t mylab.yml

# deploy a lab from mylab.yml file and regenerate all configuration artifacts
containerlab deploy -t mylab.yml --reconfigure
```
```

0 comments on commit eaa45e1

Please sign in to comment.