Add runner readiness waiting support

Now `terraform apply` will wait until the runner
is registered and ready before considering the
plan application as successful.

Author: 0x2b3bfa0

iterative/resource_runner.go

@@ -6,15 +6,19 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"log"
+	"net"
 	"os"
 	"strconv"
 	"text/template"
+	"time"
 
 	"terraform-provider-iterative/iterative/aws"
 	"terraform-provider-iterative/iterative/azure"
 	"terraform-provider-iterative/iterative/utils"
 
 	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 )
 
@@ -176,6 +180,43 @@ func resourceRunnerCreate(ctx context.Context, d *schema.ResourceData, m interfa
 		diags = resourceMachineCreate(ctx, d, m)
 	}
 
+	if diags.HasError() {
+		return diags
+	}
+
+	machineUser := "ubuntu"
+	machineKey := d.Get("ssh_private").(string)
+	machineAddress := net.JoinHostPort(d.Get("instance_ip").(string), "22")
+	machineLogCommand := "journalctl --unit cml"
+
+	var logError error
+	var logEvents string
+	err = resource.Retry(d.Timeout(schema.TimeoutCreate), func() *resource.RetryError {
+		logEvents, logError = utils.RunCommand(machineLogCommand, 10*time.Second, machineAddress, machineUser, machineKey)
+		log.Printf("[DEBUG] Collected log events: %#v", logEvents)
+		log.Printf("[DEBUG] Connection errors: %#v", logError)
+
+		if logError != nil {
+			return resource.RetryableError(fmt.Errorf("Waiting for the machine to accept connections... %s", logError))
+		} else if !utils.IsReady(logEvents) {
+			return resource.RetryableError(fmt.Errorf("Waiting for the runner to be ready... %s", logError))
+		} else {
+			return nil
+		}
+	})
+
+	if logError != nil {
+		logEvents = logError.Error()
+	}
+
+	if err != nil {
+		diags = append(diags, diag.Diagnostic{
+			Severity: diag.Error,
+			Summary:  fmt.Sprintf("Error checking the runner status"),
+			Detail:   logEvents,
+		})
+	}
+
 	return diags
 }
 

iterative/utils/runner.go

@@ -0,0 +1,41 @@
+package utils
+
+import (
+	"bufio"
+	"encoding/json"
+	"regexp"
+	"strings"
+)
+
+type LogEvent struct {
+	Level      string `json:"level"`
+	Time       string `json:"time"`
+	Repository string `json:"repo"`
+	Job        string `json:"job"`
+	Status     string `json:"status"`
+	Success    bool   `json:"success"`
+}
+
+func ParseLogEvent(logEvent string) (LogEvent, error) {
+	var result LogEvent
+	err := json.Unmarshal([]byte(logEvent), &result)
+	return result, err
+}
+
+// IsReady checks whether a runner is ready or not by
+// parsing the JSONL records from the logs it produces.
+func IsReady(logs string) bool {
+	scanner := bufio.NewScanner(strings.NewReader(logs))
+	for scanner.Scan() {
+		line := scanner.Text()
+		// Extract the JSON between curly braces from the log line.
+		record := regexp.MustCompile(`\{.+\}`).Find([]byte(line))
+		// Try to parse the retrieved JSON string into a LogEvent structure.
+		if event, err := ParseLogEvent(string(record)); err == nil {
+			if event.Status == "ready" {
+				return true
+			}
+		}
+	}
+	return false
+}

iterative/utils/runner_test.go

@@ -0,0 +1,23 @@
+package utils
+
+import "testing"
+
+func TestPositiveReadinessCheck(t *testing.T) {
+	result := IsReady(`
+        -- Logs begin at Wed 2021-01-20 00:25:37 UTC, end at Fri 2021-02-26 15:37:56 UTC. --
+        Feb 26 15:37:20 ip-172-31-6-188 cml.sh[2203]: {"level":"info","time":"···","repo":"···","status":"ready"}
+    `)
+	if !result {
+		t.Errorf("Positive readiness check")
+	}
+}
+
+func TestNegativeReadinessCheck(t *testing.T) {
+	result := IsReady(`
+        -- Logs begin at Wed 2021-01-20 00:25:37 UTC, end at Fri 2021-02-26 15:37:56 UTC. --
+        Feb 26 15:37:20 ip-172-31-6-188 cml.sh[2203]: {"level":"info","time":"···","repo":"···"}
+    `)
+	if result {
+		t.Errorf("Negative readiness check")
+	}
+}

iterative/utils/ssh.go

@@ -1,11 +1,13 @@
 package utils
 
 import (
+	"bytes"
 	"crypto/rand"
 	"crypto/rsa"
 	"crypto/x509"
 	"encoding/pem"
 	"strings"
+	"time"
 
 	"golang.org/x/crypto/ssh"
 )
@@ -42,3 +44,41 @@ func PublicFromPrivatePEM(privateKey string) (string, error) {
 
 	return pubKeyBuf.String(), nil
 }
+
+func RunCommand(command string, timeout time.Duration, hostAddress string, userName string, privateKey string) (string, error) {
+	parsedPrivateKey, err := ssh.ParsePrivateKey([]byte(privateKey))
+	if err != nil {
+		return "", err
+	}
+
+	configuration := &ssh.ClientConfig{
+		User: userName,
+		Auth: []ssh.AuthMethod{
+			ssh.PublicKeys(parsedPrivateKey),
+		},
+		HostKeyCallback: ssh.InsecureIgnoreHostKey(), // Doesn't matter for this use case, but isn't a good practice either.
+		Timeout:         timeout,
+	}
+
+	client, err := ssh.Dial("tcp", hostAddress, configuration)
+	if err != nil {
+		return "", err
+	}
+	defer client.Close()
+
+	session, err := client.NewSession()
+	if err != nil {
+		return "", err
+	}
+	defer session.Close()
+
+	var buffer bytes.Buffer
+	session.Stdout = &buffer
+	session.Stderr = &buffer
+
+	if err := session.Run(command); err == nil {
+		return buffer.String(), err
+	} else {
+		return "", err
+	}
+}