Replace default kubernetes client (#11)

* refactor: move KubernetesClient to composition of Fabric8 Client

* refactor: move from DefaultKubernetesClient to KubernetesClientBuilder

* fix: formatting

Author: ftschirpke

src/main/java/cws/k8s/scheduler/client/KubernetesClient.java renamed to src/main/java/cws/k8s/scheduler/client/CWSKubernetesClient.java

@@ -3,30 +3,48 @@
 import cws.k8s.scheduler.model.NodeWithAlloc;
 import cws.k8s.scheduler.model.PodWithAge;
 import io.fabric8.kubernetes.api.model.*;
-import io.fabric8.kubernetes.client.DefaultKubernetesClient;
-import io.fabric8.kubernetes.client.KubernetesClientException;
-import io.fabric8.kubernetes.client.Watcher;
-import io.fabric8.kubernetes.client.WatcherException;
+import io.fabric8.kubernetes.client.*;
+import io.fabric8.kubernetes.client.Config;
+import io.fabric8.kubernetes.client.dsl.MixedOperation;
+import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
+import io.fabric8.kubernetes.client.dsl.PodResource;
+import io.fabric8.kubernetes.client.dsl.Resource;
 import lombok.extern.slf4j.Slf4j;
 
 import java.math.BigDecimal;
 import java.util.*;
 
 @Slf4j
-public class KubernetesClient extends DefaultKubernetesClient  {
+public class CWSKubernetesClient {
 
-    private final Map<String, NodeWithAlloc> nodeHolder= new HashMap<>();
+    private final KubernetesClient client;
+
+    private final Map<String, NodeWithAlloc> nodeHolder = new HashMap<>();
     private final List<Informable> informables = new LinkedList<>();
 
 
-    public KubernetesClient(){
+    public CWSKubernetesClient() {
+        KubernetesClientBuilder builder = new KubernetesClientBuilder();
+        this.client = builder.build();
         for( Node node : this.nodes().list().getItems() ){
             nodeHolder.put( node.getMetadata().getName(), new NodeWithAlloc(node,this) );
         }
         this.pods().inAnyNamespace().watch( new PodWatcher( this ) );
         this.nodes().watch( new NodeWatcher( this ) );
     }
 
+    public NonNamespaceOperation<Node, NodeList, Resource<Node>> nodes() {
+        return client.nodes();
+    }
+
+    public MixedOperation<Pod, PodList, PodResource> pods() {
+        return client.pods();
+    }
+
+    public Config getConfiguration() {
+        return client.getConfiguration();
+    }
+
     public void addInformable( Informable informable ){
         synchronized ( informables ){
             informables.add( informable );
@@ -83,7 +101,7 @@ public void assignPodToNode( PodWithAge pod, String node ) {
                         .withApiVersion( nodeWithAlloc.getApiVersion() )
                         .withName( node ).endTarget()
                         .build();
-                bindings()
+                client.bindings()
                         .inNamespace( pod.getMetadata().getNamespace() )
                         .resource( build )
                         .create();
@@ -106,7 +124,7 @@ public List<NodeWithAlloc> getAllNodes(){
     }
 
     public BigDecimal getMemoryOfNode(NodeWithAlloc node ){
-        final Quantity memory = this
+        final Quantity memory = client
                 .top()
                 .nodes()
                 .metrics(node.getName())
@@ -117,9 +135,9 @@ public BigDecimal getMemoryOfNode(NodeWithAlloc node ){
 
     static class NodeWatcher implements Watcher<Node>{
 
-        private final KubernetesClient kubernetesClient;
+        private final CWSKubernetesClient kubernetesClient;
 
-        public NodeWatcher(KubernetesClient kubernetesClient) {
+        public NodeWatcher(CWSKubernetesClient kubernetesClient) {
             this.kubernetesClient = kubernetesClient;
         }
 
@@ -173,9 +191,9 @@ public void onClose(WatcherException cause) {
 
     static class PodWatcher implements Watcher<Pod> {
 
-        private final KubernetesClient kubernetesClient;
+        private final CWSKubernetesClient kubernetesClient;
 
-        public PodWatcher(KubernetesClient kubernetesClient) {
+        public PodWatcher(CWSKubernetesClient kubernetesClient) {
             this.kubernetesClient = kubernetesClient;
         }
 

src/main/java/cws/k8s/scheduler/config/Beans.java

@@ -1,6 +1,6 @@
 package cws.k8s.scheduler.config;
 
-import cws.k8s.scheduler.client.KubernetesClient;
+import cws.k8s.scheduler.client.CWSKubernetesClient;
 import lombok.AccessLevel;
 import lombok.RequiredArgsConstructor;
 import org.springframework.context.annotation.Bean;
@@ -10,12 +10,12 @@
 @RequiredArgsConstructor( access = AccessLevel.PACKAGE )
 public class Beans {
 
-    private static KubernetesClient kubernetesClient;
+    private static CWSKubernetesClient kubernetesClient;
 
     @Bean
-    KubernetesClient getClient(){
+    CWSKubernetesClient getClient(){
         if(kubernetesClient == null) {
-            kubernetesClient = new KubernetesClient();
+            kubernetesClient = new CWSKubernetesClient();
             kubernetesClient.getConfiguration().setNamespace(null);
         }
         return kubernetesClient;

src/main/java/cws/k8s/scheduler/model/NodeWithAlloc.java

@@ -1,6 +1,6 @@
 package cws.k8s.scheduler.model;
 
-import cws.k8s.scheduler.client.KubernetesClient;
+import cws.k8s.scheduler.client.CWSKubernetesClient;
 import io.fabric8.kubernetes.api.model.Node;
 import io.fabric8.kubernetes.api.model.ObjectMeta;
 import io.fabric8.kubernetes.api.model.Pod;
@@ -16,7 +16,7 @@
 @Slf4j
 public class NodeWithAlloc extends Node implements Comparable<NodeWithAlloc> {
 
-    private final transient KubernetesClient kubernetesClient;
+    private final transient CWSKubernetesClient kubernetesClient;
 
     private static final long serialVersionUID = 1L;
 
@@ -34,7 +34,7 @@ public NodeWithAlloc( String name ) {
         this.getMetadata().setName( name );
     }
 
-    public NodeWithAlloc( Node node, KubernetesClient kubernetesClient ) {
+    public NodeWithAlloc( Node node, CWSKubernetesClient kubernetesClient ) {
 
         this.kubernetesClient = kubernetesClient;
 

src/main/java/cws/k8s/scheduler/rest/SchedulerRestController.java

@@ -2,7 +2,7 @@
 
 import cws.k8s.scheduler.dag.DAG;
 import cws.k8s.scheduler.dag.InputEdge;
-import cws.k8s.scheduler.client.KubernetesClient;
+import cws.k8s.scheduler.client.CWSKubernetesClient;
 import cws.k8s.scheduler.dag.Vertex;
 import cws.k8s.scheduler.model.SchedulerConfig;
 import cws.k8s.scheduler.model.TaskConfig;
@@ -37,7 +37,7 @@
 @EnableScheduling
 public class SchedulerRestController {
 
-    private final KubernetesClient client;
+    private final CWSKubernetesClient client;
     private final boolean autoClose;
     private final ApplicationContext appContext;
     private long closedLastScheduler = -1;
@@ -50,7 +50,7 @@ public class SchedulerRestController {
     private static final Map<String, Scheduler> schedulerHolder = new HashMap<>();
 
     public SchedulerRestController(
-            @Autowired KubernetesClient client,
+            @Autowired CWSKubernetesClient client,
             @Value("#{environment.AUTOCLOSE}") String autoClose,
             @Autowired ApplicationContext appContext ){
         this.client = client;

src/main/java/cws/k8s/scheduler/scheduler/PrioritizeAssignScheduler.java

@@ -3,7 +3,7 @@
 import cws.k8s.scheduler.model.*;
 import cws.k8s.scheduler.scheduler.prioritize.Prioritize;
 import cws.k8s.scheduler.client.Informable;
-import cws.k8s.scheduler.client.KubernetesClient;
+import cws.k8s.scheduler.client.CWSKubernetesClient;
 import cws.k8s.scheduler.scheduler.nodeassign.NodeAssign;
 import cws.k8s.scheduler.util.NodeTaskAlignment;
 import lombok.extern.slf4j.Slf4j;
@@ -17,7 +17,7 @@ public class PrioritizeAssignScheduler extends Scheduler {
     private final NodeAssign nodeAssigner;
 
     public PrioritizeAssignScheduler( String execution,
-                                      KubernetesClient client,
+                                      CWSKubernetesClient client,
                                       String namespace,
                                       SchedulerConfig config,
                                       Prioritize prioritize,

src/main/java/cws/k8s/scheduler/scheduler/Scheduler.java

@@ -4,7 +4,7 @@
 import cws.k8s.scheduler.model.*;
 import cws.k8s.scheduler.util.Batch;
 import cws.k8s.scheduler.client.Informable;
-import cws.k8s.scheduler.client.KubernetesClient;
+import cws.k8s.scheduler.client.CWSKubernetesClient;
 import cws.k8s.scheduler.util.NodeTaskAlignment;
 import io.fabric8.kubernetes.api.model.Pod;
 import io.fabric8.kubernetes.client.Watch;
@@ -40,7 +40,7 @@ public abstract class Scheduler implements Informable {
     private int currentBatch = 0;
     private Batch currentBatchInstance = null;
 
-    final KubernetesClient client;
+    final CWSKubernetesClient client;
     private final Set<Task> upcomingTasks = new HashSet<>();
     private final List<Task> unscheduledTasks = new ArrayList<>( 100 );
     private final List<Task> unfinishedTasks = new ArrayList<>( 100 );
@@ -52,7 +52,7 @@ public abstract class Scheduler implements Informable {
 
     final boolean traceEnabled;
 
-    Scheduler(String execution, KubernetesClient client, String namespace, SchedulerConfig config){
+    Scheduler(String execution, CWSKubernetesClient client, String namespace, SchedulerConfig config){
         this.execution = execution;
         this.name = System.getenv( "SCHEDULER_NAME" ) + "-" + execution;
         this.namespace = namespace;