From 99f9eb166957bfa1af2ed7324f13d9cd5207a2af Mon Sep 17 00:00:00 2001
From: Andrea Peruffo <andrea.peruffo1982@gmail.com>
Date: Tue, 2 Feb 2021 17:18:52 +0000
Subject: [PATCH] Fix the label writer

---
 .../cloudflow/execution/CloudflowConfig.scala |  4 ++++
 .../execution/CloudflowConfigSpec.scala       | 23 +++++++++++++++++++
 2 files changed, 27 insertions(+)

diff --git a/tools/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/CloudflowConfig.scala b/tools/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/CloudflowConfig.scala
index e81eec4ad..c57d1e32a 100644
--- a/tools/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/CloudflowConfig.scala
+++ b/tools/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/CloudflowConfig.scala
@@ -213,6 +213,10 @@ object CloudflowConfig {
     }
   }
 
+  implicit val labelValueWriter = ConfigWriter.fromFunction[LabelValue] { label: LabelValue =>
+    ConfigValueFactory.fromAnyRef(label.value)
+  }
+
   // LabelKey
 
   final case class LabelKey(key: String)
diff --git a/tools/cloudflow-cli/src/test/scala/akka/cli/cloudflow/execution/CloudflowConfigSpec.scala b/tools/cloudflow-cli/src/test/scala/akka/cli/cloudflow/execution/CloudflowConfigSpec.scala
index 561683cfd..db2e0db46 100644
--- a/tools/cloudflow-cli/src/test/scala/akka/cli/cloudflow/execution/CloudflowConfigSpec.scala
+++ b/tools/cloudflow-cli/src/test/scala/akka/cli/cloudflow/execution/CloudflowConfigSpec.scala
@@ -718,6 +718,29 @@ class CloudflowConfigSpec extends AnyFlatSpec with Matchers with OptionValues wi
     res.failure.exception.getMessage.contains(LabelsNotAllowedOnPod) shouldBe true
   }
 
+  it should "write label values as plain strings" in {
+    // Arrange
+    val config = s"""cloudflow {
+                    |  streamlets {
+                    |    flink {
+                    |      kubernetes.pods {
+                    |        pod {
+                    |          labels: {
+                    |            mykey = myvalue
+                    |          }
+                    |        }
+                    |      }
+                    |    }
+                    |  }
+                    |}""".stripMargin
+
+    // Act
+    val res = ConfigFactory.empty().withFallback(writeConfig(loadAndValidate(ConfigSource.string(config)).get))
+
+    // Assert
+    res.getString("cloudflow.streamlets.flink.kubernetes.pods.pod.labels.mykey") shouldBe "myvalue"
+  }
+
   it should "generate proper default mounts" in {
     // Arrange
     val crFile = new File("./cloudflow-cli/src/test/resources/swiss-knife.json")