Merge branch 'release/0.34.0'

VERSION

@@ -1 +1 @@
-0.33.0
+0.34.0

backend/sources/facebook/events-router/src/main/java/co/airy/core/sources/facebook/MessageMapper.java

@@ -73,7 +73,7 @@ public List<ProducerRecord<String, SpecificRecordBase>> getRecords(Event event,
         } else if (appId != null && !appId.equals(this.facebookAppId)) {
             // Third party app
             senderId = appId;
-        } else if (appId == null && !"instagram".equals(channel.getSource())) {
+        } else if (appId == null) {
             // Sent by Facebook moderator via Facebook inbox
             senderId = getSourceConversationId(rootNode);
         } else {
@@ -107,14 +107,14 @@ public List<ProducerRecord<String, SpecificRecordBase>> getRecords(Event event,
                 .ifPresent(referralNode -> headers.put("postback.referral", referralNode.toString()));
 
         // As a content hash use the Facebook message id if present and the whole content body if not
-        final String contentId = Stream.of(message, postbackNode)
+        final Optional<String> facebookMessageId = Stream.of(message, postbackNode)
                 .filter(Objects::nonNull)
                 .findFirst()
                 .map((node) -> node.get("mid"))
-                .map(JsonNode::textValue)
-                .orElse(payload);
+                .map(JsonNode::textValue);
 
-        final String messageId = UUIDv5.fromNamespaceAndName(channel.getId(), contentId).toString();
+        final String messageId = facebookMessageId.flatMap(getMessageId)
+                .orElseGet(() -> UUIDv5.fromNamespaceAndName(channel.getId(), payload).toString());
 
         return List.of(new ProducerRecord<>(applicationCommunicationMessages, messageId, Message.newBuilder()
                 .setSource(channel.getSource())

cli/integration/version_test.go

@@ -12,7 +12,7 @@ func TestVersion(t *testing.T) {
 	}
 
 	// https://semver.org/#is-there-a-suggested-regular-expression-regex-to-check-a-semver-string
-	validVersionString := regexp.MustCompile(`^Version: (develop|(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?), GitCommit: [0-9a-f]{5,40}\n$`)
+	validVersionString := regexp.MustCompile(`Version: (develop|(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?), GitCommit: [0-9a-f]{5,40}\n$`)
 
 	if !validVersionString.MatchString(actual) {
 		t.Errorf("expected %s to be a valid version but was not", actual)

cli/pkg/cmd/root.go

@@ -9,14 +9,20 @@ import (
 	"cli/pkg/cmd/upgrade"
 	"cli/pkg/workspace"
 	"fmt"
+	"io/ioutil"
+	"net/http"
 	"os"
+	"regexp"
+	"strings"
+	"time"
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 )
 
 const cliConfigFileName = "cli.yaml"
 const cliConfigDirName = ".airy"
+const cliVersionAPI = "https://airy-core-binaries.s3.amazonaws.com/stable.txt"
 
 var cliConfigDir string
 var Version string
@@ -31,9 +37,37 @@ var RootCmd = &cobra.Command{
 		if cmd.Name() != "create" && cmd.Name() != "version" {
 			workspace.Init(cliConfigDir)
 		}
+		if !strings.Contains(Version, "alpha") {
+			cliVersion()
+		}
 	},
 }
 
+func cliVersion() {
+	latest_stable := Version
+	client := http.Client{
+		Timeout: time.Second,
+	}
+
+	resp, err := client.Get(cliVersionAPI)
+	if err != nil {
+		return
+	} else {
+		body, _ := ioutil.ReadAll(resp.Body)
+		defer resp.Body.Close()
+		temp := strings.TrimSuffix(string(body), "\n")
+		match, _ := regexp.MatchString("^[0-9]+\\.[0-9]+\\.[0-9]+$", temp)
+		if match {
+			latest_stable = temp
+		} else {
+			return
+		}
+	}
+	if Version != latest_stable {
+		fmt.Printf("Warning: Your CLI version is out of date. Please upgrade to the latest stable version: %s. \n\n", latest_stable)
+	}
+}
+
 var versionCmd = &cobra.Command{
 	Use:   "version",
 	Short: "Prints version information",

cli/pkg/helm/helm.go

@@ -1,8 +1,10 @@
 package helm
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"io/ioutil"
 
 	"github.com/airyhq/airy/infrastructure/lib/go/k8s/util"
@@ -89,6 +91,7 @@ func (h *Helm) UpgradeCharts() error {
 	chartURL := "https://airy-core-helm-charts.s3.amazonaws.com/stable/airy-" + h.version + ".tgz"
 	return h.runHelm(append([]string{"upgrade",
 		"--values", "/apps/config/airy-config-map.yaml",
+		"--namespace", h.namespace,
 		"--timeout", "10m0s",
 		"airy", chartURL}))
 }
@@ -100,6 +103,7 @@ func (h *Helm) runHelm(args []string) error {
 
 	h.cleanupJob()
 	jobsClient := h.clientset.BatchV1().Jobs(h.namespace)
+	podsClient := h.clientset.CoreV1().Pods(h.namespace)
 
 	job := &batchv1.Job{
 		ObjectMeta: metav1.ObjectMeta{
@@ -115,7 +119,7 @@ func (h *Helm) runHelm(args []string) error {
 							Name:            "helm-runner",
 							Image:           "alpine/helm:3.6.3",
 							Args:            args,
-							ImagePullPolicy: corev1.PullAlways,
+							ImagePullPolicy: corev1.PullIfNotPresent,
 							VolumeMounts: []corev1.VolumeMount{
 								{
 									Name:      "core-config",
@@ -162,14 +166,30 @@ func (h *Helm) runHelm(args []string) error {
 		switch event.Type {
 		case watch.Error:
 			return fmt.Errorf("helm run failed with error %v", event.Object)
-		case watch.Added:
 		case watch.Modified:
 			job, _ := event.Object.(*batchv1.Job)
 			if job.Status.Succeeded == 1 {
 				return nil
 			} else if job.Status.Failed == 1 {
+				opts := v1.ListOptions{
+					LabelSelector: "job-name=helm-runner",
+				}
+				pods, err := podsClient.List(context.TODO(), opts)
+				if err != nil {
+					fmt.Println(err)
+				}
+
+				for _, pod := range pods.Items {
+					fmt.Println("Logs of helm-runner pod:")
+					fmt.Println(getPodLogs(pod, h.clientset))
+				}
+
+				for _, c := range job.Status.Conditions {
+					return fmt.Errorf("Helm job failed: %s", c.Reason)
+				}
 				return fmt.Errorf("helm run failed with error %v", event.Object)
 			}
+		default:
 		}
 	}
 
@@ -214,3 +234,20 @@ func (h *Helm) cleanupJob() error {
 		PropagationPolicy: &deletionPolicy,
 	})
 }
+
+func getPodLogs(pod corev1.Pod, clientset *kubernetes.Clientset) string {
+	podLogOpts := corev1.PodLogOptions{}
+	req := clientset.CoreV1().Pods(pod.Namespace).GetLogs(pod.Name, &podLogOpts)
+	podLogs, err := req.Stream(context.TODO())
+	if err != nil {
+		return "error in opening stream"
+	}
+	defer podLogs.Close()
+
+	buf := new(bytes.Buffer)
+	_, err = io.Copy(buf, podLogs)
+	if err != nil {
+		return "error in copy information from podLogs to buf"
+	}
+	return buf.String()
+}