v0.3.3 k8s webssh

k8s/consumers.py

@@ -0,0 +1,100 @@
+from asgiref.sync import async_to_sync
+from channels.generic.websocket import WebsocketConsumer
+
+import paramiko
+import threading
+import time
+from seal import settings
+
+from channels.layers import get_channel_layer
+
+channel_layer = get_channel_layer()
+
+
+class MyThread(threading.Thread):
+    def __init__(self, chan):
+        threading.Thread.__init__(self)
+        self.chan = chan
+        self.number = 0
+
+    def run(self):
+
+        while not self.chan.chan.exit_status_ready():
+            time.sleep(0.1)
+            try:
+                data = self.chan.chan.recv(1024)
+                str_data = data.decode(encoding='utf-8')
+                if getattr(settings, 'webssh_name') in data.decode(encoding='utf-8'):
+                    self.number += 1
+
+                if "kubectl exec -it" in str_data:
+                    #不返回内容
+                    pass
+                else:
+                    if "rpc error" in str_data:
+                        async_to_sync(self.chan.channel_layer.group_send)(
+                            self.chan.scope['user'].username,
+                            {
+                                "type": "user.message",
+                                "text": "连接错误,已断开连接! 此 pod  不支持sh 或者其他未知错误!\r"
+                            },
+                        )
+                        self.chan.sshclient.close()
+                    elif self.number > 1:
+                        async_to_sync(self.chan.channel_layer.group_send)(
+                            self.chan.scope['user'].username,
+                            {
+                                "type": "user.message",
+                                "text": "程序退出,已断开连接!\r"
+                            },
+                        )
+                        self.chan.sshclient.close()
+                    else:
+                        async_to_sync(self.chan.channel_layer.group_send)(
+                            self.chan.scope['user'].username,
+                            {
+                                "type": "user.message",
+                                "text": bytes.decode(data)
+                            },
+                        )
+
+            except Exception as ex:
+                pass
+        self.chan.sshclient.close()
+        return False
+
+
+class EchoConsumer(WebsocketConsumer):
+
+    def connect(self):
+        # 创建channels group， 命名为：用户名 (最好不要中文名字)，并使用channel_layer写入到redis
+        async_to_sync(self.channel_layer.group_add)(self.scope['user'].username, self.channel_name)
+
+        self.sshclient = paramiko.SSHClient()
+        self.sshclient.load_system_host_keys()
+        self.sshclient.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        self.sshclient.connect(getattr(settings, 'webssh_ip'), getattr(settings, 'webssh_port'),
+                               getattr(settings, 'webssh_username'), getattr(settings, 'webssh_password'))
+        self.chan = self.sshclient.invoke_shell(term='xterm')
+        self.chan.settimeout(0)
+        t1 = MyThread(self)
+        t1.setDaemon(True)
+        t1.start()
+        path = self.scope['path'].split('/')
+        cmd = f"kubectl exec -it {path[2]}  -n  {path[3]}  sh  \r"
+        self.chan.send(cmd)
+
+        self.accept()
+
+    def receive(self, text_data):
+        try:
+            self.chan.send(text_data)
+        except Exception as ex:
+            pass
+            # print(str(ex))
+
+    def user_message(self, event):
+        self.send(text_data=event["text"])
+
+    def disconnect(self, close_code):
+        async_to_sync(self.channel_layer.group_discard)(self.scope['user'].username, self.channel_name)

k8s/k8sApi/core.py

@@ -1,15 +1,54 @@
 from kubernetes import client, config
 from seal import settings
 import urllib3
+from kubernetes.stream import stream
 
 
-def K8sApi():
-    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
-    configuration = client.Configuration()
-    configuration.host = getattr(settings, 'APISERVER')
-    configuration.verify_ssl = False
-    configuration.api_key = {"authorization": "Bearer " + getattr(settings, 'Token'), }
-    client.Configuration.set_default(configuration)
-    v1 = client.CoreV1Api()
+class K8sApi(object):
 
-    return v1
+    def __init__(self):
+        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
+    def get_client(self):
+        baseurl = getattr(settings, 'APISERVER')
+        token = getattr(settings, 'Token')
+        aConfiguration = client.Configuration()
+        aConfiguration.host = baseurl
+        aConfiguration.verify_ssl = False
+        aConfiguration.api_key = {"authorization": "Bearer " + token}
+        aApiClient = client.ApiClient(aConfiguration)
+        v1 = client.CoreV1Api(aApiClient)
+        return v1
+
+    def get_podlist(self):
+        client_v1 = self.get_client()
+        ret_pod = client_v1.list_pod_for_all_namespaces(watch=False)
+        return ret_pod
+
+    def get_namespacelist(self):
+        client_v1 = self.get_client()
+        ret_namespace = client_v1.list_namespace()
+        return ret_namespace
+
+    def test_pods_connect(self, podname, namespace, command, container=None):
+        client_v1 = self.get_client()
+        if stream(client_v1.connect_get_namespaced_pod_exec, podname, namespace, command=command,
+                  container=container,
+                  stderr=True, stdin=False,
+                  stdout=True, tty=False):
+            return True
+        else:
+            return False
+
+    def get_pods_exec(self, podname, namespace, command, container=None):
+        client_v1 = self.get_client()
+        if container:
+            rest = stream(client_v1.connect_get_namespaced_pod_exec, podname, namespace, command=command,
+                          container=container,
+                          stderr=True, stdin=False,
+                          stdout=True, tty=False)
+        else:
+            rest = stream(client_v1.connect_get_namespaced_pod_exec, podname, namespace, command=command,
+                          stderr=True, stdin=False,
+                          stdout=True, tty=False)
+        return rest

k8s/urls.py

@@ -4,4 +4,5 @@ app_name = "k8s"
 
 urlpatterns = [
     path('k8s-pod-list', views.K8sPodListView.as_view(), name='k8s-pod-list'),
+    path('k8s-pod-webssh', views.K8sPodWebssh.as_view(), name='k8s-pod-webssh')
 ]

k8s/views.py

@@ -10,6 +10,7 @@ from system.decorator.get_list import get_list
 from django.contrib.auth.mixins import LoginRequiredMixin, PermissionRequiredMixin
 from django.views.generic import ListView, View, DetailView, CreateView, UpdateView
 from assets.models import Ecs
+from seal import settings
 from k8s.k8sApi.core import K8sApi
 
 logger = logging.getLogger('k8s')
@@ -17,11 +18,21 @@ logger = logging.getLogger('k8s')
 
 class K8sPodListView(LoginRequiredMixin, PermissionRequiredMixin, View):
     permission_required = ('k8s.view_ecs',)
-    template_name = 'k8s/k8s-pod-list.html'
 
     def get(self, request):
-        ret = K8sApi().list_pod_for_all_namespaces(watch=False)
+        obj = K8sApi()
+        ret = obj.get_podlist()
         data = {}
         for i in ret.items:
             data[i.metadata.name] = {"ip": i.status.pod_ip, "namespace": i.metadata.namespace}
         return render(request, "k8s/k8s-pod-list.html", {"data": data})
+
+
+class K8sPodWebssh(LoginRequiredMixin, PermissionRequiredMixin, View):
+    permission_required = ('k8s.view_ecs',)
+
+    def get(self, request):
+        name = self.request.GET.get("name")
+        namespace = self.request.GET.get("namespace")
+        return render(request, "k8s/k8s-pod-webssh.html",{"name":name,"namespace":namespace})
+