feat auto update sdk

Author: liyan.90210

Changelog

@@ -1,5 +1,8 @@
 ### Change log
 
+2026-03-26 Bumped to version v1.0.219
+- Updated apis for tls/vod
+
 2026-03-19 Bumped to version v1.0.218
 - Updated apis for content_security
 

volcengine/__init__.py

@@ -1,2 +1,2 @@
 # coding:utf-8
-VERSION='v1.0.218'
+VERSION='v1.0.219'

volcengine/tls/TLSService.py

@@ -448,14 +448,15 @@ def put_logs_v2(self, request: PutLogsV2Request) -> PutLogsResponse:
             new_log = log_group.logs.add()
             new_log.time = v.time
             # 设置纳秒级时间戳（如果提供）
-            if v.time_ns is not None:
+            if v.time_ns is not None and hasattr(new_log, "TimeNs"):
                 new_log.TimeNs = v.time_ns
             for key in v.log_dict.keys():
                 log_content = new_log.contents.add()
                 log_content.key = str(key)
                 log_content.value = str(v.log_dict[key])
         put_logs_request = PutLogsRequest(request.topic_id, log_group_list,
-                                         request.hash_key, request.compression, request.content_md5)
+                                         request.hash_key, request.compression, request.content_md5,
+                                         enable_nanosecond=request.enable_nanosecond)
         return self.put_logs(put_logs_request)
 
     def describe_cursor(self, describe_cursor_request: DescribeCursorRequest) -> DescribeCursorResponse:

volcengine/tls/producer/batch_manager.py

@@ -20,7 +20,7 @@ def add_now(self, config: ProducerConfig, executor_service, client, memory_lock,
         if self.batch_log is not None:
             task = SendBatchTask(
                 self.batch_log, config, memory_lock, client, retry_queue)
-            executor_service.submit(task.run())
+            executor_service.submit(task.run)
             self.batch_log = None
 
     def remove_batch(self, batch_logs: ['BatchLog']) -> None:

volcengine/tls/producer/log_dispatcher.py

@@ -1,9 +1,10 @@
 import threading
+import time
 from concurrent.futures import ThreadPoolExecutor
 from typing import Dict
 
 from volcengine.tls.TLSService import TLSService
-from volcengine.tls.log_pb2 import LogGroup
+from volcengine.tls.log_pb2 import LogGroup, LogGroupList
 from volcengine.tls.producer.batch_manager import BatchManager
 from volcengine.tls.producer.batch_semaphore import BatchSemaphore
 from volcengine.tls.producer.producer_model import ProducerConfig, BatchLog, CallBack
@@ -104,8 +105,48 @@ def do_add(self, hash_key: str, topic_id: str, source: str, filename: str,
             raise TLSException(error_code="AddBatch Error",
                                error_message="closed LogDispatcher cannot receive logs anymore")
 
+        log_count = 0
+        earliest_log_time = None
+        latest_log_time = None
+        for log in log_group.logs:
+            if log.time <= 0:
+                now_ns = time.time_ns()
+                log.time = int(now_ns // 1_000_000)
+                if self.producer_config.enable_nanosecond and hasattr(log, "TimeNs"):
+                    try:
+                        if not log.HasField("TimeNs"):
+                            log.TimeNs = int(now_ns % 1_000_000)
+                    except ValueError:
+                        log.TimeNs = int(now_ns % 1_000_000)
+            log_count += 1
+            normalized_time = log.time
+            if log.time < 1e10:
+                normalized_time = log.time * 1000
+            elif log.time < 1e15:
+                normalized_time = log.time
+            else:
+                normalized_time = log.time // int(1e6)
+            if earliest_log_time is None or normalized_time < earliest_log_time:
+                earliest_log_time = normalized_time
+            if latest_log_time is None or normalized_time > latest_log_time:
+                latest_log_time = normalized_time
+
         # 计算批次大小
-        batch_size = len(log_group.SerializeToString())
+        def _varint_len(x: int) -> int:
+            if x < 0:
+                return 10
+            n = 1
+            while x >= (1 << 7):
+                n += 1
+                x >>= 7
+            return n
+
+        def _tag_len(field_number: int) -> int:
+            return _varint_len((field_number << 3) | 2)
+
+        group_size = log_group.ByteSize()
+        log_group_list_groups_field_number = LogGroupList.DESCRIPTOR.fields_by_name["log_groups"].number
+        batch_size = _tag_len(log_group_list_groups_field_number) + _varint_len(group_size) + group_size
         self.producer_config.check_batch_size(batch_size)
 
         # 获取内存锁
@@ -138,19 +179,21 @@ def do_add(self, hash_key: str, topic_id: str, source: str, filename: str,
 
             # 同步操作
             with batch_manager.lock:
-                self.add_to_batch_manager(batch_key, log_group, callback, batch_size, batch_manager)
+                self.add_to_batch_manager(batch_key, log_group, callback, batch_size, batch_manager,
+                                          log_count, earliest_log_time, latest_log_time)
         except Exception as e:
             # 发生异常时释放内存锁
             self.memory_lock.release(batch_size)
             raise TLSException(error_code="Add Batch Error", error_message="dispatcher add batch concurrent error")
 
     def add_to_batch_manager(self, batch_key: BatchLog.BatchKey, log_group: LogGroup,
-                             callback: CallBack, batch_size: int, batch_manager: BatchManager) -> None:
+                             callback: CallBack, batch_size: int, batch_manager: BatchManager,
+                             log_count: int, earliest_log_time: int, latest_log_time: int) -> None:
         """将日志添加到批次管理器"""
         # 尝试添加到现有批次
         batch_log = batch_manager.batch_log
         if batch_log is not None:
-            success = batch_log.try_add(log_group, batch_size, callback)
+            success = batch_log.try_add(log_group, batch_size, callback, log_count, earliest_log_time, latest_log_time)
             if success:
                 # 检查是否已满需要发送
                 if batch_manager.full_and_send_batch_request():
@@ -176,7 +219,7 @@ def add_to_batch_manager(self, batch_key: BatchLog.BatchKey, log_group: LogGroup
         batch_log = BatchLog(batch_key, self.producer_config)
         batch_manager.batch_log = batch_log
 
-        success = batch_log.try_add(log_group, batch_size, callback)
+        success = batch_log.try_add(log_group, batch_size, callback, log_count, earliest_log_time, latest_log_time)
         if not success:
             self.LOG.error(
                 f"tryAdd batchLog failed, batchKey = {str(batch_key)}, batchSize = {batch_size}, "

volcengine/tls/producer/producer.py

@@ -2,7 +2,7 @@
 import time
 from typing import Optional
 
-from volcengine.tls.log_pb2 import LogGroup
+from volcengine.tls.log_pb2 import LogGroup, LogGroupList
 from volcengine.tls.producer.batch_semaphore import BatchSemaphore
 from volcengine.tls.producer.log_dispatcher import LogDispatcher
 from volcengine.tls.producer.mover import Mover
@@ -57,49 +57,83 @@ def send_logs_v2(self, hash_key: Optional[str], topic_id: str, source: Optional[
         max_log_group_count = ProducerConfig.MAX_LOG_GROUP_COUNT
         max_log_group_size = ProducerConfig.MAX_BATCH_SIZE
 
+        def _varint_len(x: int) -> int:
+            if x < 0:
+                return 10
+            n = 1
+            while x >= (1 << 7):
+                n += 1
+                x >>= 7
+            return n
+
+        def _tag_len(field_number: int) -> int:
+            return _varint_len((field_number << 3) | 2)
+
+        log_groups_field_number = LogGroup.DESCRIPTOR.fields_by_name["logs"].number
+        log_group_list_groups_field_number = LogGroupList.DESCRIPTOR.fields_by_name["log_groups"].number
+
         log_group = LogGroup()
         if source is not None:
             log_group.source = source
         if filename is not None:
             log_group.filename = filename
 
+        base_group_size = log_group.ByteSize()
+        current_estimated_group_size = base_group_size
         current_count = 0
         for v in logs:
-            new_log = log_group.logs.add()  # pylint: disable=no-member
+            new_log = getattr(log_group, "logs").add()
             new_log.time = v.time
+            if v.time_ns is not None and hasattr(new_log, "TimeNs"):
+                new_log.TimeNs = v.time_ns
             for key in v.log_dict.keys():
                 log_content = new_log.contents.add()
                 log_content.key = str(key)
                 log_content.value = str(v.log_dict[key])
 
             current_count += 1
-            log_group_size = len(log_group.SerializeToString())
-            if log_group_size > max_log_group_size:
+            log_size = new_log.ByteSize()
+            current_estimated_group_size += _tag_len(log_groups_field_number) + _varint_len(log_size) + log_size
+            group_size = current_estimated_group_size
+            entry_size = _tag_len(log_group_list_groups_field_number) + _varint_len(group_size) + group_size
+            if entry_size > max_log_group_size:
                 if current_count == 1:
+                    actual_group_size = log_group.ByteSize()
+                    actual_entry_size = _tag_len(log_group_list_groups_field_number) + _varint_len(actual_group_size) + actual_group_size
                     raise TLSException(
                         error_code="InvalidArgument",
-                        error_message=f"log size {log_group_size} is larger than MAX_LOG_SIZE {max_log_group_size}"
+                        error_message=f"log size {actual_entry_size} is larger than MAX_LOG_SIZE {max_log_group_size}"
                     )
-                log_group.logs.pop()
+                getattr(log_group, "logs").pop()
                 current_count -= 1
+                current_estimated_group_size -= _tag_len(log_groups_field_number) + _varint_len(log_size) + log_size
                 self.dispatcher.add_batch(hash_key, topic_id, source, filename, log_group, callback)
                 log_group = LogGroup()
                 if source is not None:
                     log_group.source = source
                 if filename is not None:
                     log_group.filename = filename
-                new_log = log_group.logs.add()  # pylint: disable=no-member
+                base_group_size = log_group.ByteSize()
+                new_log = getattr(log_group, "logs").add()
                 new_log.time = v.time
+                if v.time_ns is not None and hasattr(new_log, "TimeNs"):
+                    new_log.TimeNs = v.time_ns
                 for key in v.log_dict.keys():
                     log_content = new_log.contents.add()
                     log_content.key = str(key)
                     log_content.value = str(v.log_dict[key])
                 current_count = 1
-                log_group_size = len(log_group.SerializeToString())
-                if log_group_size > max_log_group_size:
+                current_estimated_group_size = base_group_size
+                log_size = new_log.ByteSize()
+                current_estimated_group_size += _tag_len(log_groups_field_number) + _varint_len(log_size) + log_size
+                group_size = current_estimated_group_size
+                entry_size = _tag_len(log_group_list_groups_field_number) + _varint_len(group_size) + group_size
+                if entry_size > max_log_group_size:
+                    actual_group_size = log_group.ByteSize()
+                    actual_entry_size = _tag_len(log_group_list_groups_field_number) + _varint_len(actual_group_size) + actual_group_size
                     raise TLSException(
                         error_code="InvalidArgument",
-                        error_message=f"log size {log_group_size} is larger than MAX_LOG_SIZE {max_log_group_size}"
+                        error_message=f"log size {actual_entry_size} is larger than MAX_LOG_SIZE {max_log_group_size}"
                     )
 
             if current_count >= max_log_group_count:
@@ -110,6 +144,8 @@ def send_logs_v2(self, hash_key: Optional[str], topic_id: str, source: Optional[
                 if filename is not None:
                     log_group.filename = filename
                 current_count = 0
+                base_group_size = log_group.ByteSize()
+                current_estimated_group_size = base_group_size
 
         if current_count > 0:
             self.dispatcher.add_batch(hash_key, topic_id, source, filename, log_group, callback)

volcengine/tls/producer/producer_model.py

@@ -96,6 +96,7 @@ def __init__(self, endpoint: str, region: str, access_key: str, access_secret: s
         self.max_block_ms = self.DEFAULT_BLOCK_MS
         self.retry_count = self.DEFAULT_RETRY_COUNT
         self.max_reserved_attempts = self.DEFAULT_RESERVED_ATTEMPTS
+        self.enable_nanosecond = False
         self.client_config = ClientConfig(endpoint, region, access_key, access_secret, token)
 
     def __str__(self) -> str:
@@ -107,6 +108,7 @@ def __str__(self) -> str:
                 f"max_block_ms={self.max_block_ms}, "
                 f"retry_count={self.retry_count}, "
                 f"max_reserved_attempts={self.max_reserved_attempts}, "
+                f"enable_nanosecond={self.enable_nanosecond}, "
                 f"client_config={self.client_config})")
 
     def valid_config(self) -> None:
@@ -241,6 +243,8 @@ def __init__(self, batch_key: 'BatchLog.BatchKey', producer_config: ProducerConf
         self.batch_key = batch_key
         self.current_batch_size = 0
         self.current_batch_count = 0
+        self.earliest_log_time = None
+        self.latest_log_time = None
         self.call_back_list = []
         self.log_group_list = LogGroupList()
         self.producer_config = producer_config
@@ -254,13 +258,29 @@ def __init__(self, batch_key: 'BatchLog.BatchKey', producer_config: ProducerConf
         self.base_increase_backoff_ms = 1000  # 1秒
         self.LOG = get_logger(__name__)
 
-    def try_add(self, log_group: LogGroup, batch_size: int, call_back: Optional[CallBack]) -> bool:
+    def try_add(self, log_group: LogGroup, batch_size: int, call_back: Optional[CallBack],
+                log_count: Optional[int] = None, earliest_log_time: Optional[int] = None,
+                latest_log_time: Optional[int] = None) -> bool:
         """尝试添加日志组到批次中"""
         current_batch_count = self.current_batch_count
         current_batch_size = self.current_batch_size
+        if log_count is None:
+            log_count = len(log_group.logs)
+            for log in log_group.logs:
+                normalized_time = log.time
+                if log.time < 1e10:
+                    normalized_time = log.time * 1000
+                elif log.time < 1e15:
+                    normalized_time = log.time
+                else:
+                    normalized_time = log.time // int(1e6)
+                if earliest_log_time is None or normalized_time < earliest_log_time:
+                    earliest_log_time = normalized_time
+                if latest_log_time is None or normalized_time > latest_log_time:
+                    latest_log_time = normalized_time
 
         # 检查是否超过阈值
-        if (len(log_group.logs) + current_batch_count > ProducerConfig.MAX_BATCH_COUNT or
+        if (log_count + current_batch_count > ProducerConfig.MAX_BATCH_COUNT or
                 batch_size + current_batch_size > ProducerConfig.MAX_BATCH_SIZE):
             return False
 
@@ -271,8 +291,13 @@ def try_add(self, log_group: LogGroup, batch_size: int, call_back: Optional[Call
             self.call_back_list.append(call_back)
 
         # 更新当前计数
-        self.current_batch_count = current_batch_count + len(log_group.logs)
+        self.current_batch_count = current_batch_count + log_count
         self.current_batch_size = current_batch_size + batch_size
+        if log_count > 0 and earliest_log_time is not None and latest_log_time is not None:
+            if self.earliest_log_time is None or earliest_log_time < self.earliest_log_time:
+                self.earliest_log_time = earliest_log_time
+            if self.latest_log_time is None or latest_log_time > self.latest_log_time:
+                self.latest_log_time = latest_log_time
 
         return True
 

volcengine/tls/producer/send_batch_task.py

@@ -32,7 +32,18 @@ def run(self) -> None:
     def send_request(self) -> None:
         """构建并发送日志请求"""
         batch_key = self.batch_log.batch_key
-        put_logs_request = PutLogsRequest(batch_key.topic_id, self.batch_log.log_group_list, batch_key.shard_hash)
+        put_logs_request = PutLogsRequest(
+            batch_key.topic_id,
+            self.batch_log.log_group_list,
+            batch_key.shard_hash,
+            log_count=self.batch_log.current_batch_count,
+            earliest_log_time=self.batch_log.earliest_log_time,
+            latest_log_time=self.batch_log.latest_log_time,
+            enable_nanosecond=self.producer_config.enable_nanosecond,
+        )
+
+        if not self.calibrate_batch_size():
+            return
 
         try:
             put_logs_response = self.client.put_logs(put_logs_request)
@@ -45,6 +56,24 @@ def send_request(self) -> None:
 
         self.handle_success(put_logs_response)
 
+    def calibrate_batch_size(self) -> bool:
+        estimated = int(self.batch_log.current_batch_size)
+        actual = int(self.batch_log.log_group_list.ByteSize())
+        delta = actual - estimated
+        if delta > 0:
+            max_block_ms = self.producer_config.max_block_ms
+            if max_block_ms == 0:
+                self.memory_lock.acquire(delta)
+            else:
+                acquired = self.memory_lock.acquire(delta, timeout=max_block_ms / 1000)
+                if not acquired:
+                    self.handle_exception(Exception("buffer full when calibrating batch size"))
+                    return False
+        elif delta < 0:
+            self.memory_lock.release(-delta)
+        self.batch_log.current_batch_size = actual
+        return True
+
     def handle_failure(self) -> None:
         """处理失败日志"""
         self.LOG.info(f"send batch failed, batch: {self.batch_log}")

volcengine/tls/test/producer_split_unit_test.py

@@ -33,11 +33,11 @@ def test_split_by_count(self):
 
         groups = producer.dispatcher.groups
         self.assertEqual(3, len(groups))
-        self.assertEqual(10000, len(groups[0].logs))
-        self.assertEqual(10000, len(groups[1].logs))
-        self.assertEqual(5000, len(groups[2].logs))
+        self.assertEqual(10000, len(getattr(groups[0], "logs")))
+        self.assertEqual(10000, len(getattr(groups[1], "logs")))
+        self.assertEqual(5000, len(getattr(groups[2], "logs")))
         for g in groups:
-            self.assertLessEqual(len(g.logs), ProducerConfig.MAX_LOG_GROUP_COUNT)
+            self.assertLessEqual(len(getattr(g, "logs")), ProducerConfig.MAX_LOG_GROUP_COUNT)
             self.assertLessEqual(len(g.SerializeToString()), ProducerConfig.MAX_BATCH_SIZE)
 
     def test_split_by_size(self):
@@ -84,7 +84,7 @@ def test_batch_log_group_list_count_never_exceeds_32768(self):
         for _ in range(6):
             g = LogGroup()
             for _ in range(10000):
-                g.logs.append(Log())
+                getattr(g, "logs").append(Log())
             groups.append(g)
 
         batches = []