perf: 命令存储ES可根据日期动态建立索引 (#8180)

* perf: 命令存储ES可根据日期动态建立索引

* perf: 优化合并字段

* feat: 修改逻辑

apps/common/utils/timezone.py

@@ -32,6 +32,10 @@ def local_now_display(fmt='%Y-%m-%d %H:%M:%S'):
     return local_now().strftime(fmt)
 
 
+def local_now_date_display(fmt='%Y-%m-%d'):
+    return local_now().strftime(fmt)
+
+
 _rest_dt_field = DateTimeField()
 dt_parser = _rest_dt_field.to_internal_value
 dt_formatter = _rest_dt_field.to_representation

apps/locale/ja/LC_MESSAGES/django.po

@@ -30,7 +30,7 @@ msgstr "Acls"
 #: orgs/models.py:65 perms/models/base.py:83 rbac/models/role.py:29
 #: settings/models.py:29 settings/serializers/sms.py:6
 #: terminal/models/endpoint.py:10 terminal/models/endpoint.py:58
-#: terminal/models/storage.py:23 terminal/models/task.py:16
+#: terminal/models/storage.py:24 terminal/models/task.py:16
 #: terminal/models/terminal.py:100 users/forms/profile.py:32
 #: users/models/group.py:15 users/models/user.py:661
 #: xpack/plugins/cloud/models.py:28
@@ -62,7 +62,7 @@ msgstr "アクティブ"
 #: assets/models/label.py:23 ops/models/adhoc.py:38 orgs/models.py:68
 #: perms/models/base.py:93 rbac/models/role.py:37 settings/models.py:34
 #: terminal/models/endpoint.py:20 terminal/models/endpoint.py:68
-#: terminal/models/storage.py:26 terminal/models/terminal.py:114
+#: terminal/models/storage.py:27 terminal/models/terminal.py:114
 #: tickets/models/comment.py:24 tickets/models/ticket.py:154
 #: users/models/group.py:16 users/models/user.py:698
 #: xpack/plugins/change_auth_plan/models/base.py:44
@@ -305,7 +305,7 @@ msgstr "カテゴリ"
 #: assets/models/cmd_filter.py:82 assets/models/user.py:246
 #: perms/models/application_permission.py:24
 #: perms/serializers/application/user_permission.py:34
-#: terminal/models/storage.py:55 terminal/models/storage.py:119
+#: terminal/models/storage.py:56 terminal/models/storage.py:136
 #: tickets/models/flow.py:56 tickets/models/ticket.py:131
 #: tickets/serializers/ticket/meta/ticket_type/apply_application.py:29
 #: xpack/plugins/change_auth_plan/models/app.py:28
@@ -4639,7 +4639,7 @@ msgstr "オンラインセッションを持つ"
 msgid "Terminals"
 msgstr "ターミナル管理"
 
-#: terminal/backends/command/es.py:26
+#: terminal/backends/command/es.py:28
 msgid "Invalid elasticsearch config"
 msgstr "無効なElasticsearch構成"
 
@@ -4883,15 +4883,15 @@ msgstr "スレッド"
 msgid "Boot Time"
 msgstr "ブート時間"
 
-#: terminal/models/storage.py:25
+#: terminal/models/storage.py:26
 msgid "Default storage"
 msgstr "デフォルトのストレージ"
 
-#: terminal/models/storage.py:113 terminal/models/terminal.py:108
+#: terminal/models/storage.py:130 terminal/models/terminal.py:108
 msgid "Command storage"
 msgstr "コマンドストレージ"
 
-#: terminal/models/storage.py:173 terminal/models/terminal.py:109
+#: terminal/models/storage.py:190 terminal/models/terminal.py:109
 msgid "Replay storage"
 msgstr "再生ストレージ"
 
@@ -5008,14 +5008,26 @@ msgid "Port invalid"
 msgstr "ポートが無効"
 
 #: terminal/serializers/storage.py:159
+msgid "Index by date"
+msgstr "日付による索引付け"
+
+#: terminal/serializers/storage.py:160
+msgid "Whether to create an index by date"
+msgstr "現在の日付に基づいてインデックスを動的に作成するかどうか"
+
+#: terminal/serializers/storage.py:163
+msgid "Index prefix"
+msgstr "インデックス接頭辞"
+
+#: terminal/serializers/storage.py:166
 msgid "Index"
 msgstr "インデックス"
 
-#: terminal/serializers/storage.py:161
+#: terminal/serializers/storage.py:168
 msgid "Doc type"
 msgstr "Docタイプ"
 
-#: terminal/serializers/storage.py:163
+#: terminal/serializers/storage.py:170
 msgid "Ignore Certificate Verification"
 msgstr "証明書の検証を無視する"
 

apps/locale/zh/LC_MESSAGES/django.po

@@ -29,7 +29,7 @@ msgstr "访问控制"
 #: orgs/models.py:65 perms/models/base.py:83 rbac/models/role.py:29
 #: settings/models.py:29 settings/serializers/sms.py:6
 #: terminal/models/endpoint.py:10 terminal/models/endpoint.py:58
-#: terminal/models/storage.py:23 terminal/models/task.py:16
+#: terminal/models/storage.py:24 terminal/models/task.py:16
 #: terminal/models/terminal.py:100 users/forms/profile.py:32
 #: users/models/group.py:15 users/models/user.py:661
 #: xpack/plugins/cloud/models.py:28
@@ -61,7 +61,7 @@ msgstr "激活中"
 #: assets/models/label.py:23 ops/models/adhoc.py:38 orgs/models.py:68
 #: perms/models/base.py:93 rbac/models/role.py:37 settings/models.py:34
 #: terminal/models/endpoint.py:20 terminal/models/endpoint.py:68
-#: terminal/models/storage.py:26 terminal/models/terminal.py:114
+#: terminal/models/storage.py:27 terminal/models/terminal.py:114
 #: tickets/models/comment.py:24 tickets/models/ticket.py:154
 #: users/models/group.py:16 users/models/user.py:698
 #: xpack/plugins/change_auth_plan/models/base.py:44
@@ -300,7 +300,7 @@ msgstr "类别"
 #: assets/models/cmd_filter.py:82 assets/models/user.py:246
 #: perms/models/application_permission.py:24
 #: perms/serializers/application/user_permission.py:34
-#: terminal/models/storage.py:55 terminal/models/storage.py:119
+#: terminal/models/storage.py:56 terminal/models/storage.py:136
 #: tickets/models/flow.py:56 tickets/models/ticket.py:131
 #: tickets/serializers/ticket/meta/ticket_type/apply_application.py:29
 #: xpack/plugins/change_auth_plan/models/app.py:28
@@ -4565,7 +4565,7 @@ msgstr "有在线会话"
 msgid "Terminals"
 msgstr "终端管理"
 
-#: terminal/backends/command/es.py:26
+#: terminal/backends/command/es.py:28
 msgid "Invalid elasticsearch config"
 msgstr "无效的 Elasticsearch 配置"
 
@@ -4809,15 +4809,15 @@ msgstr "线程数"
 msgid "Boot Time"
 msgstr "运行时间"
 
-#: terminal/models/storage.py:25
+#: terminal/models/storage.py:26
 msgid "Default storage"
 msgstr "默认存储"
 
-#: terminal/models/storage.py:113 terminal/models/terminal.py:108
+#: terminal/models/storage.py:130 terminal/models/terminal.py:108
 msgid "Command storage"
 msgstr "命令存储"
 
-#: terminal/models/storage.py:173 terminal/models/terminal.py:109
+#: terminal/models/storage.py:190 terminal/models/terminal.py:109
 msgid "Replay storage"
 msgstr "录像存储"
 
@@ -4934,14 +4934,26 @@ msgid "Port invalid"
 msgstr "端口无效"
 
 #: terminal/serializers/storage.py:159
+msgid "Index by date"
+msgstr "按日期建索引"
+
+#: terminal/serializers/storage.py:160
+msgid "Whether to create an index by date"
+msgstr "是否根据日期动态建立索引"
+
+#: terminal/serializers/storage.py:163
+msgid "Index prefix"
+msgstr "索引前缀"
+
+#: terminal/serializers/storage.py:166
 msgid "Index"
 msgstr "索引"
 
-#: terminal/serializers/storage.py:161
+#: terminal/serializers/storage.py:168
 msgid "Doc type"
 msgstr "文档类型"
 
-#: terminal/serializers/storage.py:163
+#: terminal/serializers/storage.py:170
 msgid "Ignore Certificate Verification"
 msgstr "忽略证书认证"
 

apps/terminal/backends/command/es.py

@@ -1,11 +1,12 @@
 # -*- coding: utf-8 -*-
 #
+import pytz
+import inspect
+
 from datetime import datetime
 from functools import reduce, partial
 from itertools import groupby
-import pytz
 from uuid import UUID
-import inspect
 
 from django.utils.translation import gettext_lazy as _
 from django.db.models import QuerySet as DJQuerySet
@@ -15,6 +16,7 @@ from elasticsearch.exceptions import RequestError, NotFoundError
 
 from common.utils.common import lazyproperty
 from common.utils import get_logger
+from common.utils.timezone import local_now_date_display, utc_now
 from common.exceptions import JMSException
 from .models import AbstractSessionCommand
 
@@ -28,12 +30,13 @@ class InvalidElasticsearch(JMSException):
 
 class CommandStore(object):
     def __init__(self, config):
-        hosts = config.get("HOSTS")
-        kwargs = config.get("OTHER", {})
-        self.index = config.get("INDEX") or 'jumpserver'
         self.doc_type = config.get("DOC_TYPE") or '_doc'
+        self.index_prefix = config.get('INDEX') or 'jumpserver'
+        self.is_index_by_date = bool(config.get('INDEX_BY_DATE'))
         self.exact_fields = {}
         self.match_fields = {}
+        hosts = config.get("HOSTS")
+        kwargs = config.get("OTHER", {})
 
         ignore_verify_certs = kwargs.pop('IGNORE_VERIFY_CERTS', False)
         if ignore_verify_certs:
@@ -50,6 +53,17 @@ class CommandStore(object):
         else:
             self.match_fields.update(may_exact_fields)
 
+        self.init_index(config)
+
+    def init_index(self, config):
+        if self.is_index_by_date:
+            date = local_now_date_display()
+            self.index = '%s-%s' % (self.index_prefix, date)
+            self.query_index = '%s-alias' % self.index_prefix
+        else:
+            self.index = config.get("INDEX") or 'jumpserver'
+            self.query_index = config.get("INDEX") or 'jumpserver'
+
     def is_new_index_type(self):
         if not self.ping(timeout=3):
             return False
@@ -101,11 +115,18 @@ class CommandStore(object):
         else:
             mappings = {'mappings': {'properties': properties}}
 
+        if self.is_index_by_date:
+            mappings['aliases'] = {
+                self.query_index: {}
+            }
         try:
             self.es.indices.create(self.index, body=mappings)
             return
         except RequestError as e:
-            logger.exception(e)
+            if e.error == 'resource_already_exists_exception':
+                logger.warning(e)
+            else:
+                logger.exception(e)
 
     @staticmethod
     def make_data(command):
@@ -141,7 +162,7 @@ class CommandStore(object):
         body = self.get_query_body(**query)
 
         data = self.es.search(
-            index=self.index, doc_type=self.doc_type, body=body, from_=from_, size=size,
+            index=self.query_index, doc_type=self.doc_type, body=body, from_=from_, size=size,
             sort=sort
         )
         source_data = []
@@ -154,7 +175,7 @@ class CommandStore(object):
 
     def count(self, **query):
         body = self.get_query_body(**query)
-        data = self.es.count(index=self.index, doc_type=self.doc_type, body=body)
+        data = self.es.count(index=self.query_index, doc_type=self.doc_type, body=body)
         return data["count"]
 
     def __getattr__(self, item):

apps/terminal/models/storage.py

@@ -1,6 +1,7 @@
 from __future__ import unicode_literals
 
 import os
+
 from importlib import import_module
 
 import jms_storage
@@ -10,6 +11,7 @@ from django.conf import settings
 from common.mixins import CommonModelMixin
 from common.utils import get_logger
 from common.db.fields import EncryptJsonDictTextField
+from common.utils.timezone import local_now_date_display
 from terminal.backends import TYPE_ENGINE_MAPPING
 from .terminal import Terminal
 from .command import Command
@@ -63,6 +65,10 @@ class CommandStorage(CommonStorageModelMixin, CommonModelMixin):
     def type_server(self):
         return self.type == const.CommandStorageTypeChoices.server.value
 
+    @property
+    def type_es(self):
+        return self.type == const.CommandStorageTypeChoices.es.value
+
     @property
     def type_null_or_server(self):
         return self.type_null or self.type_server
@@ -73,6 +79,18 @@ class CommandStorage(CommonStorageModelMixin, CommonModelMixin):
         config.update({'TYPE': self.type})
         return config
 
+    @property
+    def valid_config(self):
+        config = self.config
+        if self.type_es and config.get('INDEX_BY_DATE'):
+            engine_mod = import_module(TYPE_ENGINE_MAPPING[self.type])
+            store = engine_mod.CommandStore(config)
+            store._ensure_index_exists()
+            index_prefix = config.get('INDEX') or 'jumpserver'
+            date = local_now_date_display()
+            config['INDEX'] = '%s-%s' % (index_prefix, date)
+        return config
+
     def is_valid(self):
         if self.type_null_or_server:
             return True

apps/terminal/models/terminal.py

@@ -68,7 +68,7 @@ class StorageMixin:
     def get_command_storage_config(self):
         s = self.get_command_storage()
         if s:
-            config = s.config
+            config = s.valid_config
         else:
             config = settings.DEFAULT_TERMINAL_COMMAND_STORAGE
         return config

apps/terminal/serializers/storage.py

@@ -155,6 +155,10 @@ class CommandStorageTypeESSerializer(serializers.Serializer):
         child=serializers.CharField(validators=[command_storage_es_host_format_validator]),
         label=_('Hosts'), help_text=_(hosts_help_text), allow_null=True
     )
+    INDEX_BY_DATE = serializers.BooleanField(
+        default=False, label=_('Index by date'),
+        help_text=_('Whether to create an index by date')
+    )
     INDEX = serializers.CharField(
         max_length=1024, default='jumpserver', label=_('Index'), allow_null=True
     )