Updates according to cnf-testsuite/kubectl_client#17 (#5)

Updates according to cnf-testsuite/kubectl_client#17 (#5)

Signed-off-by: Rafal Lal <[email protected]>

Author: rafal-lal

shard.lock

@@ -2,7 +2,7 @@ version: 2.0
 shards:
   cluster_tools:
     git: https://github.com/cnf-testsuite/cluster_tools.git
-    version: 1.0.0
+    version: 1.0.11
 
   docker_client:
     git: https://github.com/cnf-testsuite/docker_client.git
@@ -14,7 +14,7 @@ shards:
 
   kubectl_client:
     git: https://github.com/cnf-testsuite/kubectl_client.git
-    version: 1.0.1
+    version: 1.0.8
 
   popcorn:
     git: https://github.com/icyleaf/popcorn.git

shard.yml

@@ -20,8 +20,8 @@ dependencies:
     version: ~> 0.1.0
   kubectl_client:
     github: cnf-testsuite/kubectl_client
-    version: ~> 1.0.0
+    version: ~> 1.0.8
   cluster_tools:
     github: cnf-testsuite/cluster_tools
-    version: ~> 1.0.0
+    version: ~> 1.0.11
 

spec/kernel_introspection/k8s_spec.cr

@@ -6,20 +6,20 @@ require "../../src/kernel_introspection/k8s.cr"
 describe "KernelIntrospection::K8s" do
   before_all do
     begin
-      KubectlClient::Create.namespace("cnf-testsuite")
-    rescue e : KubectlClient::Create::AlreadyExistsError
+      KubectlClient::Apply.namespace("cnf-testsuite")
+    rescue e : KubectlClient::ShellCMD::AlreadyExistsError
     end
     ClusterTools.install
   end
 
   it "'#status_by_proc' should return all statuses for all containers in a pod", tags: ["status_by_proc"] do
-    result = KubectlClient::ShellCmd.run("kubectl run nginx --image=nginx --labels='name=nginx'", "kubectl_run_nginx", force_output=true)
+    result = KubectlClient::ShellCMD.run("kubectl run nginx --image=nginx --labels='name=nginx'")
     pods = KubectlClient::Get.pods_by_nodes(KubectlClient::Get.schedulable_nodes_list)
     pods.should_not be_nil
-    pods = KubectlClient::Get.pods_by_label(pods, "name", "nginx")
+    pods = KubectlClient::Get.pods_by_labels(pods, {"name" => "nginx"})
     pods.should_not be_nil
 
-    KubectlClient::Get.resource_wait_for_install("pod", "nginx")
+    KubectlClient::Wait.resource_wait_for_install("pod", "nginx")
     pods.size.should be > 0
     first_node = pods[0]
     statuses = KernelIntrospection::K8s.status_by_proc(first_node.dig("metadata", "name"), "nginx")
@@ -28,32 +28,30 @@ describe "KernelIntrospection::K8s" do
 
     (statuses.find{|x| x["cmdline"].includes?("nginx: master process")} ).should_not be_nil
 
-    KubectlClient::Delete.command("pod/nginx")
+    KubectlClient::Delete.resource("pod", "nginx")
   end
 
   it "'#find_first_process' should return first matching process", tags: ["find_first_proc"]  do
-    result = KubectlClient::ShellCmd.run("kubectl run nginx --image=nginx --labels='name=nginx'", "kubectl_run_nginx", force_output=true)
-    KubectlClient::Get.resource_wait_for_install("pod", "nginx")
+    result = KubectlClient::ShellCMD.run("kubectl run nginx --image=nginx --labels='name=nginx'")
+    KubectlClient::Wait.resource_wait_for_install("pod", "nginx")
     begin
-      KubectlClient::ShellCmd.run("kubectl get pods", "kubectl_get_pods",force_output=true)
       pod_info = KernelIntrospection::K8s.find_first_process("nginx: master process")
       Log.info { "pod_info: #{pod_info}"}
       (pod_info).should_not be_nil
     ensure
-      KubectlClient::Delete.command("pod/nginx")
+      KubectlClient::Delete.resource("pod", "nginx")
     end
   end
 
   it "'#find_matching_processes' should return all matching processes", tags: ["find_first_proc"]  do
-    result = KubectlClient::ShellCmd.run("kubectl run nginx --image=nginx --labels='name=nginx'", "kubectl_run_nginx", force_output=true)
-    KubectlClient::Get.resource_wait_for_install("pod", "nginx")
-    KubectlClient::ShellCmd.run("kubectl get pods", "kubectl_get_pods",force_output=true)
+    result = KubectlClient::ShellCMD.run("kubectl run nginx --image=nginx --labels='name=nginx'")
+    KubectlClient::Wait.resource_wait_for_install("pod", "nginx")
     begin
       pods_info = KernelIntrospection::K8s.find_matching_processes("nginx")
       Log.info { "pods_info: #{pods_info}"}
       (pods_info).size.should be > 0
     ensure
-      KubectlClient::Delete.command("pod/nginx")
+      KubectlClient::Delete.resource("pod", "nginx")
     end
   end
 

src/kernel_introspection/k8s.cr

@@ -3,24 +3,30 @@ require "kernel_introspection"
 module KernelIntrospection
   module K8s
     module Node
-      def self.pids(node)
+      def self.pids(node) : Array(String)
         Log.info { "pids" }
         ls_proc = ClusterTools.exec_by_node("ls /proc/", node)
+        unless ls_proc[:status].success?
+          return [] of String
+        end
         Log.info { "pids ls_proc: #{ls_proc}" }
         parsed_ls = KernelIntrospection.parse_ls(ls_proc[:output])
         pids = KernelIntrospection.pids_from_ls_proc(parsed_ls)
         pids
       end
 
-      def self.pids_by_container(container_id, node)
+      def self.pids_by_container(container_id, node) : Array(String)
         # Command explanation:
         # 1. Get all process directories in /proc with find.
-        # 2. Exec grep <container_id> for every returned process cgroup file, 
+        # 2. Exec grep <container_id> for every returned process cgroup file,
         # redirect any errors (in case some temporary process disappears/find cannot access a directory)
         # 3. Grep returns paths with desired cgroup in format /proc/<pid>/cgroup, this output gets trimmed
         # by sed to only return a list of <pid>s
         command = "/bin/sh -c \"find /proc -maxdepth 1 -regex '/proc/[0-9]+' -exec grep -l '#{container_id}' {}/cgroup \\; 2>/dev/null | sed -e 's,/proc/\\([0-9]*\\)/cgroup,\\1,'\""
         result = ClusterTools.exec_by_node(command, node)
+        unless result[:status].success?
+          return [] of String
+        end
         output = result["output"].strip
 
         pids = output.split("\n")
@@ -32,8 +38,9 @@ module KernelIntrospection
         proc_statuses = pids.map do |pid|
           Log.info { "all_statuses_by_pids pid: #{pid}" }
           proc_status = ClusterTools.exec_by_node("cat /proc/#{pid}/status", node)
-          proc_status[:output]
-        end
+          # if /proc/#{pid}/status cannot be read it means that the process is no longer available
+          proc_status[:output] if proc_status[:status].success?
+        end.compact
 
         Log.debug { "proc process_statuses_by_node: #{proc_statuses}" }
         proc_statuses
@@ -46,11 +53,11 @@ module KernelIntrospection
         status[:output]
       end
 
-      def self.cmdline_by_pid(pid : String, node) 
+      def self.cmdline_by_pid(pid : String, node)
         Log.info { "cmdline_by_pid" }
         cmdline = ClusterTools.exec_by_node("cat /proc/#{pid}/cmdline", node)
         Log.info { "cmdline_by_node cmdline: #{cmdline}" }
-        cmdline 
+        cmdline
       end
 
       def self.verify_single_proc_tree(original_parent_pid, name, proctree : Array(Hash(String, String)), excluded_processes = [] of String)
@@ -62,28 +69,27 @@ module KernelIntrospection
           status_name = "#{pt["Name"]}".strip
 
           if current_pid == original_parent_pid && ppid != "" &&
-            status_name != name
+             status_name != name
             if excluded_processes.includes?(status_name)
               next
             end
             Log.info { "top level parent (i.e. superviser -- first parent with different name): #{status_name}" }
             verified = false
-
           elsif current_pid == original_parent_pid && ppid != "" &&
-            status_name == name
+                status_name == name
             verified = verify_single_proc_tree(ppid, name, proctree, excluded_processes)
           end
         end
         Log.info { "verified?: #{verified}" }
         verified
       end
 
-      def self.proctree_by_pid(potential_parent_pid : String, node : JSON::Any, proc_statuses : (Array(String) | Nil)  = nil) : Array(Hash(String, String)) # array of status hashes
+      def self.proctree_by_pid(potential_parent_pid : String, node : JSON::Any, proc_statuses : (Array(String) | Nil) = nil) : Array(Hash(String, String)) # array of status hashes
         Log.for("proctree_by_pid").info { "proctree_by_pid potential_parent_pid: #{potential_parent_pid}" }
         proctree = [] of Hash(String, String)
         potential_parent_status : Hash(String, String) | Nil = nil
         unless proc_statuses
-          pids = pids(node) 
+          pids = pids(node)
           Log.for("proctree_by_pid").debug { "pids: #{pids}" }
           proc_statuses = all_statuses_by_pids(pids, node)
         end
@@ -92,7 +98,7 @@ module KernelIntrospection
           parsed_status = KernelIntrospection.parse_status(proc_status)
           Log.for("proctree_by_pid").debug { "parsed_status: #{parsed_status}" }
           if parsed_status
-            ppid = parsed_status["PPid"].strip 
+            ppid = parsed_status["PPid"].strip
             current_pid = parsed_status["Pid"].strip
             Log.for("proctree_by_pid").debug(&.emit(
               potential_parent_pid: potential_parent_pid,
@@ -107,9 +113,9 @@ module KernelIntrospection
                 current_pid: current_pid,
                 cmdline: cmdline
               ))
-              potential_parent_status = parsed_status.merge({"cmdline" => cmdline}) 
+              potential_parent_status = parsed_status.merge({"cmdline" => cmdline})
               proctree << potential_parent_status
-            # Add descendants of the parent pid
+              # Add descendants of the parent pid
             elsif ppid == potential_parent_pid && ppid != current_pid
               Log.for("proctree_by_pid").debug(&.emit(
                 "proctree_by_pid ppid == pid && ppid != current_pid",
@@ -131,25 +137,37 @@ module KernelIntrospection
       end
     end
 
-    def self.proc(pod_name, container_name, namespace : String | Nil = nil)
+    def self.proc(pod_name, container_name, namespace : String?) : Array(Int32)
       Log.info { "proc namespace: #{namespace}" }
-      # todo if container_name nil, dont use container (assume one container)
-      resp = KubectlClient.exec("-ti #{pod_name} --container #{container_name} -- ls /proc/", namespace: namespace)
-      KernelIntrospection.parse_proc(resp[:output].to_s)
+      begin
+        resp = KubectlClient::Utils.exec(pod_name.to_s, "ls /proc/", container_name: container_name, namespace: namespace)
+      rescue ex : KubectlClient::ShellCMD::K8sClientCMDException
+        Log.warn { "Exception rescued: #{ex.message}" }
+        return [] of Int32
+      end
+      KernelIntrospection.parse_proc(resp[:output])
     end
 
     def self.cmdline(pod_name, container_name, pid, namespace : String | Nil = nil)
       Log.info { "cmdline namespace: #{namespace}" }
-      # todo if container_name nil, dont use container (assume one container)
-      resp = KubectlClient.exec("-ti #{pod_name} --container #{container_name} -- cat /proc/#{pid}/cmdline", namespace: namespace)
-      resp[:output].to_s.strip
+      begin
+        resp = KubectlClient::Utils.exec(pod_name.to_s, "cat /proc/#{pid}/cmdline", container_name: container_name, namespace: namespace)
+      rescue ex : KubectlClient::ShellCMD::K8sClientCMDException
+        Log.warn { "Exception rescued: #{ex.message}" }
+        return ""
+      end
+      resp[:output].strip
     end
 
     def self.status(pod_name, container_name, pid, namespace : String | Nil = nil)
-      # todo if container_name nil, dont use container (assume one container)
       Log.info { "status namespace: #{namespace}" }
-      resp = KubectlClient.exec("-ti #{pod_name} --container #{container_name} -- cat /proc/#{pid}/status", namespace: namespace)
-      KernelIntrospection.parse_status(resp[:output].to_s)
+      begin
+        resp = KubectlClient::Utils.exec(pod_name.to_s, "cat /proc/#{pid}/status", container_name: container_name, namespace: namespace)
+      rescue ex : KubectlClient::ShellCMD::K8sClientCMDException
+        Log.warn { "Exception rescued: #{ex.message}" }
+        return nil
+      end
+      KernelIntrospection.parse_status(resp[:output])
     end
 
     def self.status_by_proc(pod_name, container_name, namespace : String | Nil = nil)
@@ -160,15 +178,13 @@ module KernelIntrospection
       }.compact
     end
 
-
     alias MatchingProcessInfo = NamedTuple(
       node: JSON::Any,
       pod: JSON::Any,
-      container_status: JSON::Any, 
+      container_status: JSON::Any,
       status: String,
       pid: String,
-      cmdline: String
-    )
+      cmdline: String)
 
     # #todo overload with regex
     def self.find_first_process(process_name) : (MatchingProcessInfo | Nil)
@@ -180,7 +196,7 @@ module KernelIntrospection
         pods.map do |pod|
           status = pod["status"]
           if status["containerStatuses"]?
-              container_statuses = status["containerStatuses"].as_a
+            container_statuses = status["containerStatuses"].as_a
             Log.debug { "container_statuses: #{container_statuses}" }
             container_statuses.map do |container_status|
               ready = container_status.dig("ready").as_bool
@@ -200,7 +216,7 @@ module KernelIntrospection
               if process[:output] =~ /#{process_name}/
                 ret = {node: node, pod: pod, container_status: container_status, status: status[:output], pid: pid.to_s, cmdline: process[:output]}
                 Log.for("find_first_process").info { "status found: #{ret}" }
-                break 
+                break
               end
             end
           end
@@ -241,14 +257,13 @@ module KernelIntrospection
                 result = {node: node, pod: pod, container_status: container_status, status: status[:output], pid: pid.to_s, cmdline: process[:output]}
                 results.push(result)
                 Log.for("find_matching_processes").info { "status found: #{result}" }
-                # break 
+                # break
               end
             end
           end
         end
       end
       results
     end
-
   end
 end