feat: fast no-sync write support

Author: jansimonb

influxdb_client_3/__init__.py

@@ -7,10 +7,10 @@
 from pyarrow import ArrowException
 
 from influxdb_client_3.exceptions import InfluxDB3ClientQueryError
+from influxdb_client_3.exceptions import InfluxDBError
 from influxdb_client_3.query.query_api import QueryApi as _QueryApi, QueryApiOptionsBuilder
 from influxdb_client_3.read_file import UploadFile
 from influxdb_client_3.write_client import InfluxDBClient as _InfluxDBClient, WriteOptions, Point
-from influxdb_client_3.exceptions import InfluxDBError
 from influxdb_client_3.write_client.client.write_api import WriteApi as _WriteApi, SYNCHRONOUS, ASYNCHRONOUS, \
     PointSettings, DefaultWriteOptions, WriteType
 from influxdb_client_3.write_client.domain.write_precision import WritePrecision
@@ -24,6 +24,7 @@
 INFLUX_PRECISION = "INFLUX_PRECISION"
 INFLUX_AUTH_SCHEME = "INFLUX_AUTH_SCHEME"
 INFLUX_GZIP_THRESHOLD = "INFLUX_GZIP_THRESHOLD"
+INFLUX_WRITE_NO_SYNC = "INFLUX_WRITE_NO_SYNC"
 
 
 def write_client_options(**kwargs):
@@ -112,9 +113,15 @@ def _parse_precision(precision):
     :rtype: WritePrecision
     :raises ValueError: If the provided precision is not valid.
     """
-    if precision not in [WritePrecision.NS, WritePrecision.MS, WritePrecision.S, WritePrecision.US]:
-        raise ValueError(f"Invalid precision value: {precision}")
-    return precision
+    if precision == WritePrecision.NS or precision == "nanosecond":
+        return WritePrecision.NS
+    if precision == WritePrecision.US or precision == "microsecond":
+        return WritePrecision.US
+    if precision == WritePrecision.MS or precision == "millisecond":
+        return WritePrecision.MS
+    if precision == WritePrecision.S or precision == "second":
+        return WritePrecision.S
+    raise ValueError(f"Invalid precision value: {precision}")
 
 
 def _parse_gzip_threshold(threshold):
@@ -141,6 +148,26 @@ def _parse_gzip_threshold(threshold):
     return threshold
 
 
+def _parse_write_no_sync(write_no_sync):
+    """
+    Parses and validates the provided write no sync value.
+
+    This function ensures that the given value is a valid boolean,
+    and it raises an appropriate error if the value is not valid.
+
+    :param write_no_sync: The input value to be parsed and validated.
+    :type write_no_sync: Any
+    :return: The validated write no sync value as an boolean.
+    :rtype: bool
+    :raises ValueError: If the provided value is not a boolean.
+    """
+    try:
+        write_no_sync = bool(write_no_sync)
+    except (TypeError, ValueError):
+        raise ValueError(f"Invalid write no sync value: {write_no_sync}. Must be boolean.")
+    return write_no_sync
+
+
 class InfluxDBClient3:
     def __init__(
             self,
@@ -286,6 +313,10 @@ def from_env(cls, **kwargs: Any) -> 'InfluxDBClient3':
             kwargs['gzip_threshold'] = _parse_gzip_threshold(gzip_threshold)
             kwargs['enable_gzip'] = True
 
+        write_no_sync = os.getenv(INFLUX_WRITE_NO_SYNC)
+        if write_no_sync is not None:
+            write_options.no_sync = _parse_write_no_sync(write_no_sync)
+
         precision = os.getenv(INFLUX_PRECISION)
         if precision is not None:
             write_options.write_precision = _parse_precision(precision)

influxdb_client_3/write_client/client/write/point.py

@@ -13,6 +13,7 @@
 EPOCH = datetime.fromtimestamp(0, tz=timezone.utc)
 
 DEFAULT_WRITE_PRECISION = WritePrecision.NS
+DEFAULT_WRITE_NO_SYNC = False
 
 _ESCAPE_MEASUREMENT = str.maketrans({
     ',': r'\,',

influxdb_client_3/write_client/client/write_api.py

@@ -19,7 +19,7 @@
 from influxdb_client_3.write_client.client._base import _BaseWriteApi, _HAS_DATACLASS
 from influxdb_client_3.write_client.client.util.helpers import get_org_query_param
 from influxdb_client_3.write_client.client.write.dataframe_serializer import DataframeSerializer
-from influxdb_client_3.write_client.client.write.point import Point, DEFAULT_WRITE_PRECISION
+from influxdb_client_3.write_client.client.write.point import Point, DEFAULT_WRITE_PRECISION, DEFAULT_WRITE_NO_SYNC
 from influxdb_client_3.write_client.client.write.retry import WritesRetry
 from influxdb_client_3.write_client.domain import WritePrecision
 from influxdb_client_3.write_client.rest import _UTF_8_encoding
@@ -41,7 +41,8 @@ class WriteType(Enum):
 
 class DefaultWriteOptions(Enum):
     write_type = WriteType.synchronous
-    write_precision = WritePrecision.NS
+    write_precision = DEFAULT_WRITE_PRECISION
+    no_sync = DEFAULT_WRITE_NO_SYNC
 
 
 class WriteOptions(object):
@@ -57,6 +58,7 @@ def __init__(self, write_type: WriteType = WriteType.batching,
                  exponential_base=2,
                  max_close_wait=300_000,
                  write_precision=DEFAULT_WRITE_PRECISION,
+                 no_sync=DEFAULT_WRITE_NO_SYNC,
                  write_scheduler=ThreadPoolScheduler(max_workers=1)) -> None:
         """
         Create write api configuration.
@@ -72,7 +74,9 @@ def __init__(self, write_type: WriteType = WriteType.batching,
         :param max_retry_delay: the maximum delay between each retry attempt in milliseconds
         :param max_retry_time: total timeout for all retry attempts in milliseconds, if 0 retry is disabled
         :param exponential_base: base for the exponential retry delay
-        :parama max_close_wait: the maximum time to wait for writes to be flushed if close() is called
+        :param max_close_wait: the maximum time to wait for writes to be flushed if close() is called
+        :param write_precision: precision to use when writing points to InfluxDB
+        :param no_sync: skip waiting for WAL persistence on write
         :param write_scheduler:
         """
         self.write_type = write_type
@@ -87,6 +91,7 @@ def __init__(self, write_type: WriteType = WriteType.batching,
         self.write_scheduler = write_scheduler
         self.max_close_wait = max_close_wait
         self.write_precision = write_precision
+        self.no_sync = no_sync
 
     def to_retry_strategy(self, **kwargs):
         """
@@ -375,14 +380,16 @@ def write(self, bucket: str, org: str = None,
             return self._write_batching(bucket, org, record,
                                         write_precision, **kwargs)
 
+        no_sync = self._write_options.no_sync
+
         payloads = defaultdict(list)
         self._serialize(record, write_precision, payloads, **kwargs)
 
         _async_req = True if self._write_options.write_type == WriteType.asynchronous else False
 
         def write_payload(payload):
             final_string = b'\n'.join(payload[1])
-            return self._post_write(_async_req, bucket, org, final_string, payload[0])
+            return self._post_write(_async_req, bucket, org, final_string, payload[0], no_sync)
 
         results = list(map(write_payload, payloads.items()))
         if not _async_req:
@@ -519,18 +526,21 @@ def _retry_callback_delegate(exception):
         else:
             _retry_callback_delegate = None
 
+        no_sync = self._write_options.no_sync
+
         retry = self._write_options.to_retry_strategy(retry_callback=_retry_callback_delegate)
 
         self._post_write(False, batch_item.key.bucket, batch_item.key.org, batch_item.data,
-                         batch_item.key.precision, urlopen_kw={'retries': retry})
+                         batch_item.key.precision, no_sync, urlopen_kw={'retries': retry})
 
         logger.debug("Write request finished %s", batch_item)
 
         return _BatchResponse(data=batch_item)
 
-    def _post_write(self, _async_req, bucket, org, body, precision, **kwargs):
+    def _post_write(self, _async_req, bucket, org, body, precision, no_sync, **kwargs):
 
         return self._write_service.post_write(org=org, bucket=bucket, body=body, precision=precision,
+                                              no_sync=no_sync,
                                               async_req=_async_req,
                                               content_type="text/plain; charset=utf-8",
                                               **kwargs)

influxdb_client_3/write_client/domain/write_precision_converter.py

@@ -0,0 +1,30 @@
+from influxdb_client_3.write_client.domain import WritePrecision
+
+
+class WritePrecisionConverter(object):
+
+    @staticmethod
+    def to_v2_api_string(precision):
+        """
+        Converts WritePrecision to its string representation for V2 API.
+        """
+        if precision in [WritePrecision.NS, WritePrecision.US, WritePrecision.MS, WritePrecision.S]:
+            return precision
+        else:
+            raise ValueError("Unsupported precision '%s'" % precision)
+
+    @staticmethod
+    def to_v3_api_string(precision):
+        """
+        Converts WritePrecision to its string representation for V3 API.
+        """
+        if precision == WritePrecision.NS:
+            return "nanosecond"
+        elif precision == WritePrecision.US:
+            return "microsecond"
+        elif precision == WritePrecision.MS:
+            return "millisecond"
+        elif precision == WritePrecision.S:
+            return "second"
+        else:
+            raise ValueError("Unsupported precision '%s'" % precision)