k8s: Use contexts instead of namespaces for flexibility

.travis.yml

@@ -30,7 +30,7 @@ services:
 branches:
   only:
     - master
-    - 0.9.x
+    - /0\.[0-9]+\.x/
 
 install:
   - bin/install-promtool

core/src/main/resources/nelson/defaults.cfg

@@ -239,7 +239,6 @@ nelson {
   #       scheduler {
   #         scheduler = "nomad"
   #         kubernetes {
-  #           in-cluster = false
   #           timeout    = 10 seconds
   #           kubeconfig = "/opt/application/conf/kubeconfig"
   #         }

core/src/main/scala/Config.scala

@@ -17,7 +17,6 @@
 package nelson
 
 import nelson.BannedClientsConfig.HttpUserAgent
-import nelson.Infrastructure.KubernetesMode
 import nelson.audit.{Auditor,AuditEvent}
 import nelson.cleanup.ExpirationPolicy
 import nelson.docker.Docker
@@ -542,14 +541,10 @@ object Config {
         })
     }
 
-    def readKubernetesOutClusterParams(kfg: KConfig): Option[KubernetesMode] =
-      kfg.lookup[String]("kubeconfig").map(kubeconfig => KubernetesMode.OutCluster(Paths.get(kubeconfig)))
-
     def readKubernetesInfrastructure(kfg: KConfig): Option[Infrastructure.Kubernetes] = for {
-      inCluster <- kfg.lookup[Boolean]("in-cluster")
-      mode      <- if (inCluster) Some(KubernetesMode.InCluster) else readKubernetesOutClusterParams(kfg)
-      timeout   <- kfg.lookup[FiniteDuration]("timeout")
-    } yield Infrastructure.Kubernetes(mode, timeout)
+      kubeconfig <- kfg.lookup[String]("kubeconfig").map(k => Paths.get(k))
+      timeout    <- kfg.lookup[FiniteDuration]("timeout")
+    } yield Infrastructure.Kubernetes(kubeconfig, timeout)
 
     def readNomadScheduler(kfg: KConfig): IO[SchedulerOp ~> IO] =
       readNomadInfrastructure(kfg) match {
@@ -614,8 +609,8 @@ object Config {
 
         case Some("kubernetes") =>
           readKubernetesInfrastructure(schedConfig.subconfig("kubernetes")) match {
-            case Some(Infrastructure.Kubernetes(mode, timeout)) =>
-              val kubectl = new Kubectl(mode)
+            case Some(Infrastructure.Kubernetes(kubeconfig, timeout)) =>
+              val kubectl = new Kubectl(kubeconfig)
               IO.pure((
                 new KubernetesShell(kubectl, timeout, ec, schedulerPool),
                 new KubernetesHealthClient(kubectl, timeout, ec, schedulerPool),

core/src/main/scala/Datacenter.scala

@@ -73,22 +73,10 @@ object Infrastructure {
   )
 
   final case class Kubernetes(
-    mode: KubernetesMode,
+    kubeconfig: Path,
     timeout: FiniteDuration
   )
 
-  sealed abstract class KubernetesMode extends Product with Serializable {
-    def environment: List[(String, String)] = this match {
-      case KubernetesMode.InCluster => List.empty
-      case KubernetesMode.OutCluster(path) => List(("KUBECONFIG", path.toString))
-    }
-  }
-
-  object KubernetesMode {
-    final case object InCluster extends KubernetesMode
-    final case class OutCluster(kubeconfig: Path) extends KubernetesMode
-  }
-
   final case class Credentials(
     username: String,
     password: String

core/src/main/scala/Kubectl.scala

@@ -1,7 +1,6 @@
 package nelson
 
 import nelson.Datacenter.StackName
-import nelson.Infrastructure.KubernetesMode
 import nelson.health.{HealthCheck, HealthStatus, Passing, Failing, Unknown}
 
 import argonaut.{CursorHistory, DecodeJson, Parse}
@@ -14,25 +13,26 @@ import fs2.Stream
 
 import java.io.{ByteArrayInputStream, InputStream}
 import java.nio.charset.StandardCharsets.UTF_8
+import java.nio.file.Path
 
 import scala.sys.process.{Process, ProcessLogger}
 import scala.collection.mutable.ListBuffer
 
-final class Kubectl(mode: KubernetesMode) {
+final class Kubectl(kubeconfig: Path) {
   import Kubectl._
 
-  def apply(payload: String): IO[String] = {
+  def apply(payload: String, namespace: NamespaceName): IO[String] = {
     val input = IO { new ByteArrayInputStream(payload.getBytes(UTF_8)) }
     for {
-      result <- exec(List("kubectl", "apply", "-f", "-"), input)
+      result <- exec(List("kubectl", "apply", "--context", namespace.root.asString, "-f", "-"), input)
       output <- result.output
     } yield output.mkString("\n")
   }
 
-  def delete(payload: String): IO[String] = {
+  def delete(payload: String, namespace: NamespaceName): IO[String] = {
     val input = IO { new ByteArrayInputStream(payload.getBytes(UTF_8)) }
     for {
-      result <- exec(List("kubectl", "delete", "-f", "-"), input)
+      result <- exec(List("kubectl", "delete", "--context", namespace.root.asString, "-f", "-"), input)
       output <- result.output
     } yield output.mkString("\n")
   }
@@ -53,7 +53,7 @@ final class Kubectl(mode: KubernetesMode) {
 
   def getPods(namespace: NamespaceName, stackName: StackName): IO[List[HealthStatus]] = {
     implicit val healthStatusDecoder = healthStatusDecodeJson
-    exec(List("kubectl", "get", "pods", "-l", s"stackName=${stackName.toString}", "-n", namespace.root.asString, "-o", "json"), emptyStdin)
+    exec(List("kubectl", "get", "pods", "-l", s"stackName=${stackName.toString}", "--context", namespace.root.asString, "-o", "json"), emptyStdin)
       .flatMap(_.output)
       .flatMap { stdout =>
         IO.fromEither(for {
@@ -64,14 +64,14 @@ final class Kubectl(mode: KubernetesMode) {
   }
 
   def getDeployment(namespace: NamespaceName, stackName: StackName): IO[DeploymentStatus] =
-    exec(List("kubectl", "get", "deployment", stackName.toString, "-n", namespace.root.asString, "-o", "json"), emptyStdin)
+    exec(List("kubectl", "get", "deployment", stackName.toString, "--context", namespace.root.asString, "-o", "json"), emptyStdin)
       .flatMap(_.output)
       .flatMap { stdout =>
         IO.fromEither(Parse.decodeEither[DeploymentStatus](stdout.mkString("\n")).leftMap(kubectlJsonError))
       }
 
   def getCronJob(namespace: NamespaceName, stackName: StackName): IO[JobStatus] =
-    exec(List("kubectl", "get", "job", "-l", s"stackName=${stackName.toString}", "-n", namespace.root.asString, "-o", "json"), emptyStdin)
+    exec(List("kubectl", "get", "job", "-l", s"stackName=${stackName.toString}", "--context", namespace.root.asString, "-o", "json"), emptyStdin)
       .flatMap(_.output)
       .flatMap { stdout =>
         IO.fromEither(for {
@@ -81,14 +81,14 @@ final class Kubectl(mode: KubernetesMode) {
       }
 
   def getJob(namespace: NamespaceName, stackName: StackName): IO[JobStatus] =
-    exec(List("kubectl", "get", "job", stackName.toString, "-n", namespace.root.asString, "-o", "json"), emptyStdin)
+    exec(List("kubectl", "get", "job", stackName.toString, "--context", namespace.root.asString, "-o", "json"), emptyStdin)
       .flatMap(_.output)
       .flatMap { stdout =>
         IO.fromEither(Parse.decodeEither[JobStatus](stdout.mkString("\n")).leftMap(kubectlJsonError))
       }
 
   private def deleteV1(namespace: String, objectType: String, name: String): IO[Output] =
-    exec(List("kubectl", "delete", objectType, name, "-n", namespace), emptyStdin)
+    exec(List("kubectl", "delete", objectType, name, "--context", namespace), emptyStdin)
 
   private def exec(cmd: List[String], stdin: IO[InputStream]): IO[Output] = {
     // We need the new cats-effect resource safety hotness..
@@ -98,7 +98,7 @@ final class Kubectl(mode: KubernetesMode) {
           stdout <- IO(ListBuffer.empty[String])
           stderr <- IO(ListBuffer.empty[String])
           logger <- IO(ProcessLogger(sout => { stdout += sout; () }, serr => { stderr += serr; () }))
-          exitCode <- IO((Process(cmd, None, mode.environment: _*) #< is).run(logger).exitValue)
+          exitCode <- IO((Process(cmd, None, ("KUBECONFIG", kubeconfig.toString)) #< is).run(logger).exitValue)
         } yield Output(stdout.toList, stderr.toList, exitCode)
       }
     }, is => IO(is.close()))

core/src/main/scala/scheduler/KubernetesShell.scala

@@ -58,7 +58,7 @@ final class KubernetesShell(
         }
       }
 
-    deployment.renderedBlueprint.fold(fallback)(spec => kubectl.delete(spec).void)
+    deployment.renderedBlueprint.fold(fallback)(spec => kubectl.delete(spec, ns).void)
   }
 
   // Janky heuristic to see if an attempted (legacy) deletion failed because
@@ -89,7 +89,7 @@ final class KubernetesShell(
 
     for {
       t <- template
-      r <- kubectl.apply(t.render(env))
+      r <- kubectl.apply(t.render(env), ns)
     } yield r
   }
 

core/src/test/resources/nelson/datacenters-missing-consul.cfg

@@ -32,7 +32,7 @@ nelson {
         scheduler {
           scheduler = "kubernetes"
           kubernetes {
-            in-cluster = true
+            kubeconfig = "~/.kube/config"
             timeout = 1 second
           }
         }
@@ -68,7 +68,7 @@ nelson {
         scheduler {
           scheduler = "kubernetes"
           kubernetes {
-            in-cluster = true
+            kubeconfig = /foo/bar/.kube/config
             timeout = 1 second
           }
         }

core/src/test/resources/nelson/datacenters-missing-domain.cfg

@@ -31,7 +31,7 @@ nelson {
         scheduler {
           scheduler = "kubernetes"
           kubernetes {
-            in-cluster = true
+            kubeconfig = "~/.kube/config"
             timeout = 1 second
           }
         }

core/src/test/resources/nelson/datacenters.cfg

@@ -31,7 +31,7 @@ nelson {
         scheduler {
           scheduler = "kubernetes"
           kubernetes {
-            in-cluster = true
+            kubeconfig = "~/.kube/config"
             timeout = 1 second
           }
         }
@@ -77,7 +77,6 @@ nelson {
         scheduler {
           scheduler = "kubernetes"
           kubernetes {
-            in-cluster = false
             kubeconfig = "/foo/bar/.kube/config"
             timeout = 2 seconds
           }

docs/src/hugo/content/getting-started/install.md

@@ -186,13 +186,10 @@ nelson {
         scheduler {
           scheduler = "kubernetes"
           kubernetes {
-            # is nelson hosted on the cluster it is managing?
-            in-cluster = false
-            # in the event that Nelson is running on a substrate
-            # outside of the deployment clusters, specify the path
-            # to the kubeconfig which will dictate how kubectl does
-            # authentication with the cluster and so forth.
+            # specify the path to the kubeconfig which will dictate
+            # kubectl do authentication with the cluster and so forth.
             kubeconfig = /path/to/.kube/config
+            timeout = 10 seconds
           }
         }
 

etc/development/http/http.dev.cfg

@@ -93,7 +93,7 @@ nelson {
         scheduler {
           scheduler = "kubernetes"
           kubernetes {
-            in-cluster = true
+            kubeconfig = "~/.kube/config"
             timeout = 1 second
           }
         }

project.sbt

@@ -27,4 +27,4 @@ lazy val http = project.dependsOn(core % "test->test;compile->compile")
 
 enablePlugins(DisablePublishingPlugin)
 
-addCommandAlias("ci", ";test;coverageReport;coverageAggregate;tut;unidoc")
+addCommandAlias("ci", ";test;coverageReport;coverageAggregate")