chore(quickwit): add send app log to quickwit

Author: duanhongyi

charts/controller/templates/_helpers.tpl

@@ -106,6 +106,14 @@ env:
 - name: WORKFLOW_MANAGER_SECRET_KEY
   value: "{{ .Values.workflowManagerSecretKey }}"
 {{- end }}
+- name: POD_IP
+  valueFrom:
+    fieldRef:
+      fieldPath: status.podIP
+- name: POD_NAME
+  valueFrom:
+    fieldRef:
+      fieldPath: metadata.name
 - name: WORKFLOW_NAMESPACE
   valueFrom:
     fieldRef:
@@ -154,6 +162,8 @@ env:
       name: controller-creds
       key: passport-secret
 {{- end }}
+- name: QUICKWIT_INDEXER_URL
+  value: http://drycc-quickwit-indexer:7280
 - name: QUICKWIT_SEARCHER_URL
   value: http://drycc-quickwit-searcher:7280
 - name: QUICKWIT_LOG_INDEX_PREFIX

charts/controller/templates/controller-celery-deloyment.yaml

@@ -47,7 +47,7 @@ spec:
         args:
           - /bin/bash
           - -c
-          - celery --app api worker --queues controller.low,controller.middle,controller.high --autoscale=32,1 --loglevel=WARNING
+          - celery --app api worker --queues controller.low,controller.high,controller.middle --autoscale=32,1 --loglevel=WARNING
         {{- end }}
         {{- with index .Values "celery" "resources" }}
         resources:

rootfs/api/filer.py

@@ -5,7 +5,8 @@
 from django.core.cache import cache
 from requests.auth import HTTPBasicAuth
 
-from .utils import random_string, get_session, CacheLock
+from .tasks import send_app_log
+from .utils import random_string, get_httpclient, CacheLock
 
 logger = logging.getLogger(__name__)
 
@@ -21,6 +22,7 @@ def __init__(self, app_id, volume, scheduler):
 
     def log(self, message, level=logging.INFO):
         logger.log(level, "[{}]: {}".format(self.app_id, message))
+        send_app_log.delay(self.app_id, message, level)
 
     @property
     def server(self):
@@ -90,7 +92,8 @@ def request(self, method, server, path="/", **kwargs):
         cache.touch(self.cache_key, timeout=settings.DRYCC_FILER_DURATION)
         url = f"http://{server["address"]}:{self.bind.split(":")[1]}/{path}"
         kwargs["auth"] = HTTPBasicAuth(server["username"], server["password"])
-        return get_session().request(method, url, **kwargs)
+        with get_httpclient() as session:
+            session.request(method, url, **kwargs)
 
     def get(self, path, **kwargs):
         return self.request("GET", self.server, path, **kwargs)

rootfs/api/models/app.py

@@ -20,7 +20,8 @@
 from django.contrib.auth import get_user_model
 from rest_framework.exceptions import ValidationError
 
-from api.utils import get_session, dict_diff
+from api.tasks import send_app_log
+from api.utils import get_httpclient, dict_diff
 from api.exceptions import AlreadyExists, DryccException, ServiceUnavailable
 from api.utils import (
     CacheLock, DeployLock, generate_app_name, apply_tasks, validate_reserved_names)
@@ -185,6 +186,7 @@ def log(self, message, level=logging.INFO):
         as "belonging" to the application instead of the controller and will be handled
         accordingly.
         """
+        send_app_log.delay(self.id, message, level)
         logger.log(level, "[{}]: {}".format(self.id, message))
 
     def create(self, *args, **kwargs):  # noqa
@@ -1007,9 +1009,10 @@ def _verify_http_health(self, service, **kwargs):
         response = None
         for _ in range(10):
             try:
-                # http://docs.python-requests.org/en/master/user/advanced/#timeouts
-                response = get_session().get(url, timeout=req_timeout)
-                failed = False
+                with get_httpclient() as session:
+                    # http://docs.python-requests.org/en/master/user/advanced/#timeouts
+                    response = session.get(url, timeout=req_timeout)
+                    failed = False
             except requests.exceptions.RequestException:
                 # In case of a failure where response object is not available
                 failed = True

rootfs/api/models/appsettings.py

@@ -6,6 +6,7 @@
 
 from rest_framework.exceptions import NotFound
 from django.contrib.auth import get_user_model
+from api.tasks import send_app_log
 from api.utils import dict_diff
 from api.exceptions import DryccException, AlreadyExists, UnprocessableEntity
 from .base import UuidAuditedModel
@@ -48,6 +49,7 @@ def log(self, message, level=logging.INFO):
         as "belonging" to the application instead of the controller and will be handled
         accordingly.
         """
+        send_app_log.delay(self.app.id, message, level)
         logger.log(level, "[{}]: {}".format(self.app.id, message))
 
     def previous(self):

rootfs/api/models/base.py

@@ -11,7 +11,7 @@
 from django.utils.translation import gettext_lazy as _
 
 
-from api.utils import get_session, validate_json, get_scheduler
+from api.utils import get_httpclient, validate_json, get_scheduler
 
 token_manager_oauth_schema = {
     "$schema": "http://json-schema.org/schema#",
@@ -49,18 +49,18 @@ class Meta:
         abstract = True
 
     def app_log(self, app_id, msg):
-        session = get_session()
-        session.post(
-            "",
-            json={
-                "timestamp": int(time.time()),
-                "log": msg,
-                "kubernetes": {
-                    "namespace": app_id,
-                    "container_name": "drycc-controller",
+        with get_httpclient() as session:
+            session.post(
+                "",
+                json={
+                    "timestamp": int(time.time()),
+                    "log": msg,
+                    "kubernetes": {
+                        "namespace": app_id,
+                        "container_name": "drycc-controller",
+                    }
                 }
-            }
-        )
+            )
 
     @property
     def scheduler(self):

rootfs/api/models/gateway.py

@@ -6,6 +6,7 @@
 from django.db.models import Q
 from django.http import Http404
 
+from api.tasks import send_app_log
 from api.utils import validate_label
 from api.exceptions import ServiceUnavailable
 from scheduler import KubeException
@@ -35,6 +36,7 @@ def log(self, message, level=logging.INFO):
         as "belonging" to the application instead of the controller and will be handled
         accordingly.
         """
+        send_app_log.delay(self.app.id, message, level)
         logger.log(level, "[{}]: {}".format(self.app.id, message))
 
     def add(self, port, protocol):
@@ -255,6 +257,7 @@ def log(self, message, level=logging.INFO):
         as "belonging" to the application instead of the controller and will be handled
         accordingly.
         """
+        send_app_log.delay(self.app.id, message, level)
         logger.log(level, "[{}]: {}".format(self.app.id, message))
 
     def refresh_to_k8s(self):

rootfs/api/models/release.py

@@ -7,7 +7,7 @@
 from django.db.models import Q
 from django.contrib.auth import get_user_model
 from django.db.models import F, Func, Value, JSONField
-from api.tasks import run_pipeline
+from api.tasks import run_pipeline, send_app_log
 from api.exceptions import DryccException, AlreadyExists
 from scheduler import KubeHTTPException
 from scheduler.resources.pod import DEFAULT_CONTAINER_PORT
@@ -130,6 +130,7 @@ def log(self, message, level=logging.INFO):
         as "belonging" to the application instead of the controller and will be handled
         accordingly.
         """
+        send_app_log.delay(self.app.id, message, level)
         logger.log(level, "[{}]: {}".format(self.app.id, message))
 
     def new(self, user, config, build, summary=None):

rootfs/api/models/resource.py

@@ -3,6 +3,7 @@
 from django.db import models, transaction
 from django.contrib.auth import get_user_model
 from api.exceptions import DryccException, AlreadyExists, ServiceUnavailable
+from api.tasks import send_app_log
 from api.utils import validate_label, get_scheduler
 from scheduler import KubeException
 from .base import UuidAuditedModel
@@ -124,6 +125,7 @@ def log(self, message, level=logging.INFO):
         as "belonging" to the application instead of the controller and will be handled
         accordingly.
         """
+        send_app_log.delay(self.app.id, message, level)
         logger.log(level, "[{}]: {}".format(self.app.id, message))
 
     def bind(self, *args, **kwargs):

rootfs/api/models/service.py

@@ -2,6 +2,7 @@
 from functools import partial
 from django.db import models
 from django.contrib.auth import get_user_model
+from api.tasks import send_app_log
 from api.utils import validate_json
 from api.exceptions import ServiceUnavailable
 from scheduler import KubeException
@@ -131,6 +132,7 @@ def log(self, message, level=logging.INFO):
         as "belonging" to the application instead of the controller and will be handled
         accordingly.
         """
+        send_app_log.delay(self.app.id, message, level)
         logger.log(level, "[{}]: {}".format(self.app.id, message))
 
     def refresh_k8s_svc(self):