openrvdas.logger.writers.influxdb_writer

No module-level documentation available.
  1#!/usr/bin/env python3
  2
  3import time
  4import logging
  5from typing import Union
  6try:
  7    import urllib3
  8    URLLIB3_INSTALLED = True
  9except ImportError:
 10    URLLIB3_INSTALLED = False
 11
 12
 13from logger.utils.das_record import DASRecord  # noqa: E402
 14from logger.writers.writer import Writer  # noqa: E402
 15
 16INFLUXDB_AUTH_TOKEN = INFLUXDB_ORG = INFLUXDB_URL = INFLUXDB_BUCKET = None
 17try:
 18    from database.influxdb.settings import INFLUXDB_AUTH_TOKEN, INFLUXDB_ORG  # noqa: E402
 19    from database.influxdb.settings import INFLUXDB_BUCKET  # noqa: E402
 20    from database.influxdb.settings import INFLUXDB_URL, INFLUXDB_VERIFY_SSL  # noqa: E402
 21    INFLUXDB_SETTINGS_FOUND = True
 22except (ModuleNotFoundError, ImportError):
 23    INFLUXDB_SETTINGS_FOUND = False
 24    INFLUXDB_VERIFY_SSL = False
 25
 26try:
 27    from influxdb_client import InfluxDBClient  # noqa: E402
 28    from influxdb_client.client.write_api import ASYNCHRONOUS  # noqa: E402
 29    INFLUXDB_CLIENT_FOUND = True
 30except (ModuleNotFoundError, ImportError):
 31    INFLUXDB_CLIENT_FOUND = False
 32
 33
 34class InfluxDBWriter(Writer):
 35    """Write to the specified file. If filename is empty, write to stdout."""
 36
 37    def __init__(self, bucket_name=INFLUXDB_BUCKET, measurement_name=None,
 38                 tags=None, auth_token=INFLUXDB_AUTH_TOKEN,
 39                 org=INFLUXDB_ORG, url=INFLUXDB_URL,
 40                 verify_ssl=INFLUXDB_VERIFY_SSL, **kwargs):
 41        """Write data records to the InfluxDB.
 42        ```
 43        bucket_name - the name of the bucket in InfluxDB.  If the bucket does
 44                  not exists then this writer will try to create it.
 45
 46        measurement_name - optional measurement name to use. If not provided,
 47                  writer will use the record's data_id.
 48
 49        tags - optional tags to be applied to records submitted to InfluxDB
 50               API.
 51
 52               Example:
 53               tags:
 54                   tag0: value0
 55                   tag1:
 56                       value: value1
 57                       filter:
 58                           - measurement1
 59                           - measurement2
 60                       default: defaultValue1
 61                   tag2:
 62                       value: value2
 63                       filter: measurement2
 64
 65        auth_token - The auth token required by the InfluxDB instance. If omitted,
 66                  will look for value in imported INFLUXDB_AUTH_TOKEN and throw
 67                  an exception if it is not found.
 68
 69        org - The organization to associate with in the InfluxDB
 70                  instance. If omitted, will look for value in imported
 71                  INFLUXDB_ORG and throw an exception if it is not found.
 72
 73        url - The URL at which to connect with the InfluxDB instance. If
 74                  omitted, will look for value in imported INFLUXDB_ORG
 75                  and throw an exception if it is not found.
 76
 77        verify_ssl - If the URL begins with 'https', SSL will be used for the
 78                  connection. If so, and verify_ssl is true, the writer will
 79                  attempt to verify the validity of the relevant SSL certificate.
 80        ```
 81        """
 82        super().__init__(**kwargs)  # processes 'quiet' and type hints
 83
 84        if not URLLIB3_INSTALLED:
 85            raise ImportError('InfluxDBWriter requires Python "urllib3" module; '
 86                              'please run "pip install urllib3"')
 87        if not auth_token:
 88            raise RuntimeError('No auth token specified in InfluxDBWriter and '
 89                               'none found in database/influxdb/settings.py. Have '
 90                               'you copied over database/influxdb/settings.py.dist '
 91                               'to database/influxdb/settings.py and followed the '
 92                               'configuration instructions in it?')
 93        if not org:
 94            raise RuntimeError('No organization specified in InfluxDBWriter and '
 95                               'none found in database/influxdb/settings.py. Have '
 96                               'you copied over database/influxdb/settings.py.dist '
 97                               'to database/influxdb/settings.py and followed the '
 98                               'configuration instructions in it?')
 99        if not url:
100            raise RuntimeError('No URL specified in InfluxDBWriter and '
101                               'none found in database/influxdb/settings.py. Have '
102                               'you copied over database/influxdb/settings.py.dist '
103                               'to database/influxdb/settings.py and followed the '
104                               'configuration instructions in it?')
105
106        if not INFLUXDB_SETTINGS_FOUND:
107            raise RuntimeError('File database/influxdb/settings.py not found. '
108                               'InfluxDB functionality is not available. Have '
109                               'you copied over database/influxdb/settings.py.dist '
110                               'to database/influxdb/settings.py and followed the '
111                               'configuration instructions in it?')
112        if not INFLUXDB_CLIENT_FOUND:
113            raise RuntimeError('Python module influxdb_client not found. Please '
114                               'install using "pip install influxdb_client" prior '
115                               'to using InfluxDBWriter.')
116
117        if tags and not isinstance(tags, dict):
118            raise RuntimeError('The specified tags kwarg must be None or a dict')
119
120        self.tags = {'*': {}}
121        if tags:
122            for tag, details in tags.items():
123                if isinstance(details, str):
124                    self.tags['*'][tag] = details
125
126                if isinstance(details, dict) and 'filter' in details:
127                    if isinstance(details['filter'], str):
128                        details['filter'] = [details['filter']]
129
130                    if 'default' in details:
131                        self.tags['*'][tag] = details['default']
132
133                    for filter_item in details['filter']:
134                        if filter_item not in self.tags:
135                            self.tags[filter_item] = {}
136
137                        self.tags[filter_item][tag] = details['value']
138
139        self.auth_token = auth_token
140        self.org = org
141        self.url = url
142        self.use_ssl = url.find('https:') == 0
143        self.verify_ssl = verify_ssl
144        self.bucket_name = bucket_name
145        self.measurement_name = measurement_name
146        self.write_api = None
147
148        # If we've chosen not to verify SSL, urllib3 will complain
149        # mightily in the logs each time we make a call.
150        urllib3.disable_warnings()
151
152        # TODO: retry connecting if connection dies while writing.
153        self._connect()
154
155    ############################
156    def _connect(self):
157
158        while not self.write_api:
159            client = InfluxDBClient(url=self.url, token=self.auth_token, org=self.org,
160                                    ssl=self.use_ssl, verify_ssl=self.verify_ssl)  # type: ignore
161            # get the orgID from the name:
162            try:
163                organizations_api = client.organizations_api()
164                orgs = organizations_api.find_organizations()
165            except BaseException:
166                self.client = None
167                logging.warning('Error connecting to the InfluxDB API. '
168                                'Please confirm that InfluxDB is running and '
169                                'that the authentication token is correct.'
170                                'Sleeping before trying again.')
171                time.sleep(5)
172                continue
173
174            # Look up the organization id for our org
175            our_org = next((org for org in orgs if org.name == self.org), None)
176            if not our_org:
177                logging.fatal('Can not find org "%s" in InfluxDB', self.org)
178                raise RuntimeError('Can not find org "%s" in InfluxDB' % self.org)
179            self.org_id = our_org.id
180
181            # get the bucketID from the name:
182            bucket_api = client.buckets_api()
183            bucket = bucket_api.find_bucket_by_name(self.bucket_name)
184
185            # if the bucket does not exist then try to create it
186            if bucket:
187                self.bucket_id = bucket.id
188            else:
189                try:
190                    logging.info('Creating new bucket for: %s', self.bucket_name)
191                    new_bucket = bucket_api.create_bucket(bucket_name=self.bucket_name,
192                                                          org_id=self.org_id)
193                    self.bucket_id = new_bucket.id
194                except BaseException:
195                    logging.fatal('Can not create InfluxDB bucket "%s"', self.bucket_name)
196                    raise RuntimeError('Can not create InfluxDB bucket "%s"'
197                                       % self.bucket_name)
198
199            self.write_api = client.write_api(write_options=ASYNCHRONOUS)
200
201    ############################
202    def write(self, record: Union[DASRecord, dict]):
203        """Note: Assume record is a dict or DASRecord or list of
204        dict/DASRecord. In each record look for 'fields', 'data_id' and
205        'timestamp' (UTC epoch seconds). If data_id is missing, use the
206        bucket_name we were initialized with.
207        """
208
209        def record_to_influx(record):
210            """Put a single record into the format that InfluxDB wants."""
211            if isinstance(record, DASRecord):
212                data_id = record.data_id
213                fields = record.fields
214                timestamp = record.timestamp
215            else:
216                data_id = record.get('data_id')
217                fields = record.get('fields', {})
218                timestamp = record.get('timestamp') or time.time()
219
220            measurement = self.measurement_name or data_id
221            tags = {**{'sensor': measurement}, **self.tags['*']}
222
223            if measurement in self.tags:
224                tags = {**tags, **self.tags[measurement]}
225
226            influxDB_record = {
227                'measurement': self.measurement_name or data_id,
228                'tags': tags,
229                'fields': fields,
230                'time': int(timestamp * 1000000000)
231            }
232            return influxDB_record
233
234        # See if it's something we can process, and if not, try digesting
235        if not self.can_process_record(record):  # inherited from BaseModule()
236            self.digest_record(record)  # inherited from BaseModule()
237            return
238
239        try:
240            logging.debug('InfluxDBWriter writing record: %s', record)
241            influxDB_record = record_to_influx(record)
242            # logging.info('influxdb\n bucket: %s\nrecord: %s',
243            #             self.bucket_name, pprint.pformat(influxDB_record))
244            self.write_api.write(self.bucket_id, self.org_id, influxDB_record)
245
246        except Exception as e:
247            if not self.quiet:
248                logging.warning('InfluxDBWriter exception: %s', str(e))
249                logging.warning('InfluxDBWriter could not ingest record '
250                                'type %s: %s', type(record), str(record))
class InfluxDBWriter(logger.writers.writer.Writer): 
 35class InfluxDBWriter(Writer):
 36    """Write to the specified file. If filename is empty, write to stdout."""
 37
 38    def __init__(self, bucket_name=INFLUXDB_BUCKET, measurement_name=None,
 39                 tags=None, auth_token=INFLUXDB_AUTH_TOKEN,
 40                 org=INFLUXDB_ORG, url=INFLUXDB_URL,
 41                 verify_ssl=INFLUXDB_VERIFY_SSL, **kwargs):
 42        """Write data records to the InfluxDB.
 43        ```
 44        bucket_name - the name of the bucket in InfluxDB.  If the bucket does
 45                  not exists then this writer will try to create it.
 46
 47        measurement_name - optional measurement name to use. If not provided,
 48                  writer will use the record's data_id.
 49
 50        tags - optional tags to be applied to records submitted to InfluxDB
 51               API.
 52
 53               Example:
 54               tags:
 55                   tag0: value0
 56                   tag1:
 57                       value: value1
 58                       filter:
 59                           - measurement1
 60                           - measurement2
 61                       default: defaultValue1
 62                   tag2:
 63                       value: value2
 64                       filter: measurement2
 65
 66        auth_token - The auth token required by the InfluxDB instance. If omitted,
 67                  will look for value in imported INFLUXDB_AUTH_TOKEN and throw
 68                  an exception if it is not found.
 69
 70        org - The organization to associate with in the InfluxDB
 71                  instance. If omitted, will look for value in imported
 72                  INFLUXDB_ORG and throw an exception if it is not found.
 73
 74        url - The URL at which to connect with the InfluxDB instance. If
 75                  omitted, will look for value in imported INFLUXDB_ORG
 76                  and throw an exception if it is not found.
 77
 78        verify_ssl - If the URL begins with 'https', SSL will be used for the
 79                  connection. If so, and verify_ssl is true, the writer will
 80                  attempt to verify the validity of the relevant SSL certificate.
 81        ```
 82        """
 83        super().__init__(**kwargs)  # processes 'quiet' and type hints
 84
 85        if not URLLIB3_INSTALLED:
 86            raise ImportError('InfluxDBWriter requires Python "urllib3" module; '
 87                              'please run "pip install urllib3"')
 88        if not auth_token:
 89            raise RuntimeError('No auth token specified in InfluxDBWriter and '
 90                               'none found in database/influxdb/settings.py. Have '
 91                               'you copied over database/influxdb/settings.py.dist '
 92                               'to database/influxdb/settings.py and followed the '
 93                               'configuration instructions in it?')
 94        if not org:
 95            raise RuntimeError('No organization specified in InfluxDBWriter and '
 96                               'none found in database/influxdb/settings.py. Have '
 97                               'you copied over database/influxdb/settings.py.dist '
 98                               'to database/influxdb/settings.py and followed the '
 99                               'configuration instructions in it?')
100        if not url:
101            raise RuntimeError('No URL specified in InfluxDBWriter and '
102                               'none found in database/influxdb/settings.py. Have '
103                               'you copied over database/influxdb/settings.py.dist '
104                               'to database/influxdb/settings.py and followed the '
105                               'configuration instructions in it?')
106
107        if not INFLUXDB_SETTINGS_FOUND:
108            raise RuntimeError('File database/influxdb/settings.py not found. '
109                               'InfluxDB functionality is not available. Have '
110                               'you copied over database/influxdb/settings.py.dist '
111                               'to database/influxdb/settings.py and followed the '
112                               'configuration instructions in it?')
113        if not INFLUXDB_CLIENT_FOUND:
114            raise RuntimeError('Python module influxdb_client not found. Please '
115                               'install using "pip install influxdb_client" prior '
116                               'to using InfluxDBWriter.')
117
118        if tags and not isinstance(tags, dict):
119            raise RuntimeError('The specified tags kwarg must be None or a dict')
120
121        self.tags = {'*': {}}
122        if tags:
123            for tag, details in tags.items():
124                if isinstance(details, str):
125                    self.tags['*'][tag] = details
126
127                if isinstance(details, dict) and 'filter' in details:
128                    if isinstance(details['filter'], str):
129                        details['filter'] = [details['filter']]
130
131                    if 'default' in details:
132                        self.tags['*'][tag] = details['default']
133
134                    for filter_item in details['filter']:
135                        if filter_item not in self.tags:
136                            self.tags[filter_item] = {}
137
138                        self.tags[filter_item][tag] = details['value']
139
140        self.auth_token = auth_token
141        self.org = org
142        self.url = url
143        self.use_ssl = url.find('https:') == 0
144        self.verify_ssl = verify_ssl
145        self.bucket_name = bucket_name
146        self.measurement_name = measurement_name
147        self.write_api = None
148
149        # If we've chosen not to verify SSL, urllib3 will complain
150        # mightily in the logs each time we make a call.
151        urllib3.disable_warnings()
152
153        # TODO: retry connecting if connection dies while writing.
154        self._connect()
155
156    ############################
157    def _connect(self):
158
159        while not self.write_api:
160            client = InfluxDBClient(url=self.url, token=self.auth_token, org=self.org,
161                                    ssl=self.use_ssl, verify_ssl=self.verify_ssl)  # type: ignore
162            # get the orgID from the name:
163            try:
164                organizations_api = client.organizations_api()
165                orgs = organizations_api.find_organizations()
166            except BaseException:
167                self.client = None
168                logging.warning('Error connecting to the InfluxDB API. '
169                                'Please confirm that InfluxDB is running and '
170                                'that the authentication token is correct.'
171                                'Sleeping before trying again.')
172                time.sleep(5)
173                continue
174
175            # Look up the organization id for our org
176            our_org = next((org for org in orgs if org.name == self.org), None)
177            if not our_org:
178                logging.fatal('Can not find org "%s" in InfluxDB', self.org)
179                raise RuntimeError('Can not find org "%s" in InfluxDB' % self.org)
180            self.org_id = our_org.id
181
182            # get the bucketID from the name:
183            bucket_api = client.buckets_api()
184            bucket = bucket_api.find_bucket_by_name(self.bucket_name)
185
186            # if the bucket does not exist then try to create it
187            if bucket:
188                self.bucket_id = bucket.id
189            else:
190                try:
191                    logging.info('Creating new bucket for: %s', self.bucket_name)
192                    new_bucket = bucket_api.create_bucket(bucket_name=self.bucket_name,
193                                                          org_id=self.org_id)
194                    self.bucket_id = new_bucket.id
195                except BaseException:
196                    logging.fatal('Can not create InfluxDB bucket "%s"', self.bucket_name)
197                    raise RuntimeError('Can not create InfluxDB bucket "%s"'
198                                       % self.bucket_name)
199
200            self.write_api = client.write_api(write_options=ASYNCHRONOUS)
201
202    ############################
203    def write(self, record: Union[DASRecord, dict]):
204        """Note: Assume record is a dict or DASRecord or list of
205        dict/DASRecord. In each record look for 'fields', 'data_id' and
206        'timestamp' (UTC epoch seconds). If data_id is missing, use the
207        bucket_name we were initialized with.
208        """
209
210        def record_to_influx(record):
211            """Put a single record into the format that InfluxDB wants."""
212            if isinstance(record, DASRecord):
213                data_id = record.data_id
214                fields = record.fields
215                timestamp = record.timestamp
216            else:
217                data_id = record.get('data_id')
218                fields = record.get('fields', {})
219                timestamp = record.get('timestamp') or time.time()
220
221            measurement = self.measurement_name or data_id
222            tags = {**{'sensor': measurement}, **self.tags['*']}
223
224            if measurement in self.tags:
225                tags = {**tags, **self.tags[measurement]}
226
227            influxDB_record = {
228                'measurement': self.measurement_name or data_id,
229                'tags': tags,
230                'fields': fields,
231                'time': int(timestamp * 1000000000)
232            }
233            return influxDB_record
234
235        # See if it's something we can process, and if not, try digesting
236        if not self.can_process_record(record):  # inherited from BaseModule()
237            self.digest_record(record)  # inherited from BaseModule()
238            return
239
240        try:
241            logging.debug('InfluxDBWriter writing record: %s', record)
242            influxDB_record = record_to_influx(record)
243            # logging.info('influxdb\n bucket: %s\nrecord: %s',
244            #             self.bucket_name, pprint.pformat(influxDB_record))
245            self.write_api.write(self.bucket_id, self.org_id, influxDB_record)
246
247        except Exception as e:
248            if not self.quiet:
249                logging.warning('InfluxDBWriter exception: %s', str(e))
250                logging.warning('InfluxDBWriter could not ingest record '
251                                'type %s: %s', type(record), str(record))

Write to the specified file. If filename is empty, write to stdout.

InfluxDBWriter( bucket_name=None, measurement_name=None, tags=None, auth_token=None, org=None, url=None, verify_ssl=False, **kwargs) 
 38    def __init__(self, bucket_name=INFLUXDB_BUCKET, measurement_name=None,
 39                 tags=None, auth_token=INFLUXDB_AUTH_TOKEN,
 40                 org=INFLUXDB_ORG, url=INFLUXDB_URL,
 41                 verify_ssl=INFLUXDB_VERIFY_SSL, **kwargs):
 42        """Write data records to the InfluxDB.
 43        ```
 44        bucket_name - the name of the bucket in InfluxDB.  If the bucket does
 45                  not exists then this writer will try to create it.
 46
 47        measurement_name - optional measurement name to use. If not provided,
 48                  writer will use the record's data_id.
 49
 50        tags - optional tags to be applied to records submitted to InfluxDB
 51               API.
 52
 53               Example:
 54               tags:
 55                   tag0: value0
 56                   tag1:
 57                       value: value1
 58                       filter:
 59                           - measurement1
 60                           - measurement2
 61                       default: defaultValue1
 62                   tag2:
 63                       value: value2
 64                       filter: measurement2
 65
 66        auth_token - The auth token required by the InfluxDB instance. If omitted,
 67                  will look for value in imported INFLUXDB_AUTH_TOKEN and throw
 68                  an exception if it is not found.
 69
 70        org - The organization to associate with in the InfluxDB
 71                  instance. If omitted, will look for value in imported
 72                  INFLUXDB_ORG and throw an exception if it is not found.
 73
 74        url - The URL at which to connect with the InfluxDB instance. If
 75                  omitted, will look for value in imported INFLUXDB_ORG
 76                  and throw an exception if it is not found.
 77
 78        verify_ssl - If the URL begins with 'https', SSL will be used for the
 79                  connection. If so, and verify_ssl is true, the writer will
 80                  attempt to verify the validity of the relevant SSL certificate.
 81        ```
 82        """
 83        super().__init__(**kwargs)  # processes 'quiet' and type hints
 84
 85        if not URLLIB3_INSTALLED:
 86            raise ImportError('InfluxDBWriter requires Python "urllib3" module; '
 87                              'please run "pip install urllib3"')
 88        if not auth_token:
 89            raise RuntimeError('No auth token specified in InfluxDBWriter and '
 90                               'none found in database/influxdb/settings.py. Have '
 91                               'you copied over database/influxdb/settings.py.dist '
 92                               'to database/influxdb/settings.py and followed the '
 93                               'configuration instructions in it?')
 94        if not org:
 95            raise RuntimeError('No organization specified in InfluxDBWriter and '
 96                               'none found in database/influxdb/settings.py. Have '
 97                               'you copied over database/influxdb/settings.py.dist '
 98                               'to database/influxdb/settings.py and followed the '
 99                               'configuration instructions in it?')
100        if not url:
101            raise RuntimeError('No URL specified in InfluxDBWriter and '
102                               'none found in database/influxdb/settings.py. Have '
103                               'you copied over database/influxdb/settings.py.dist '
104                               'to database/influxdb/settings.py and followed the '
105                               'configuration instructions in it?')
106
107        if not INFLUXDB_SETTINGS_FOUND:
108            raise RuntimeError('File database/influxdb/settings.py not found. '
109                               'InfluxDB functionality is not available. Have '
110                               'you copied over database/influxdb/settings.py.dist '
111                               'to database/influxdb/settings.py and followed the '
112                               'configuration instructions in it?')
113        if not INFLUXDB_CLIENT_FOUND:
114            raise RuntimeError('Python module influxdb_client not found. Please '
115                               'install using "pip install influxdb_client" prior '
116                               'to using InfluxDBWriter.')
117
118        if tags and not isinstance(tags, dict):
119            raise RuntimeError('The specified tags kwarg must be None or a dict')
120
121        self.tags = {'*': {}}
122        if tags:
123            for tag, details in tags.items():
124                if isinstance(details, str):
125                    self.tags['*'][tag] = details
126
127                if isinstance(details, dict) and 'filter' in details:
128                    if isinstance(details['filter'], str):
129                        details['filter'] = [details['filter']]
130
131                    if 'default' in details:
132                        self.tags['*'][tag] = details['default']
133
134                    for filter_item in details['filter']:
135                        if filter_item not in self.tags:
136                            self.tags[filter_item] = {}
137
138                        self.tags[filter_item][tag] = details['value']
139
140        self.auth_token = auth_token
141        self.org = org
142        self.url = url
143        self.use_ssl = url.find('https:') == 0
144        self.verify_ssl = verify_ssl
145        self.bucket_name = bucket_name
146        self.measurement_name = measurement_name
147        self.write_api = None
148
149        # If we've chosen not to verify SSL, urllib3 will complain
150        # mightily in the logs each time we make a call.
151        urllib3.disable_warnings()
152
153        # TODO: retry connecting if connection dies while writing.
154        self._connect()

Write data records to the InfluxDB.

bucket_name - the name of the bucket in InfluxDB.  If the bucket does
          not exists then this writer will try to create it.

measurement_name - optional measurement name to use. If not provided,
          writer will use the record's data_id.

tags - optional tags to be applied to records submitted to InfluxDB
       API.

       Example:
       tags:
           tag0: value0
           tag1:
               value: value1
               filter:
                   - measurement1
                   - measurement2
               default: defaultValue1
           tag2:
               value: value2
               filter: measurement2

auth_token - The auth token required by the InfluxDB instance. If omitted,
          will look for value in imported INFLUXDB_AUTH_TOKEN and throw
          an exception if it is not found.

org - The organization to associate with in the InfluxDB
          instance. If omitted, will look for value in imported
          INFLUXDB_ORG and throw an exception if it is not found.

url - The URL at which to connect with the InfluxDB instance. If
          omitted, will look for value in imported INFLUXDB_ORG
          and throw an exception if it is not found.

verify_ssl - If the URL begins with 'https', SSL will be used for the
          connection. If so, and verify_ssl is true, the writer will
          attempt to verify the validity of the relevant SSL certificate.
tags
auth_token
org
url
use_ssl
verify_ssl
bucket_name
measurement_name
write_api
def write(self, record: Union[logger.utils.das_record.DASRecord, dict]): 
203    def write(self, record: Union[DASRecord, dict]):
204        """Note: Assume record is a dict or DASRecord or list of
205        dict/DASRecord. In each record look for 'fields', 'data_id' and
206        'timestamp' (UTC epoch seconds). If data_id is missing, use the
207        bucket_name we were initialized with.
208        """
209
210        def record_to_influx(record):
211            """Put a single record into the format that InfluxDB wants."""
212            if isinstance(record, DASRecord):
213                data_id = record.data_id
214                fields = record.fields
215                timestamp = record.timestamp
216            else:
217                data_id = record.get('data_id')
218                fields = record.get('fields', {})
219                timestamp = record.get('timestamp') or time.time()
220
221            measurement = self.measurement_name or data_id
222            tags = {**{'sensor': measurement}, **self.tags['*']}
223
224            if measurement in self.tags:
225                tags = {**tags, **self.tags[measurement]}
226
227            influxDB_record = {
228                'measurement': self.measurement_name or data_id,
229                'tags': tags,
230                'fields': fields,
231                'time': int(timestamp * 1000000000)
232            }
233            return influxDB_record
234
235        # See if it's something we can process, and if not, try digesting
236        if not self.can_process_record(record):  # inherited from BaseModule()
237            self.digest_record(record)  # inherited from BaseModule()
238            return
239
240        try:
241            logging.debug('InfluxDBWriter writing record: %s', record)
242            influxDB_record = record_to_influx(record)
243            # logging.info('influxdb\n bucket: %s\nrecord: %s',
244            #             self.bucket_name, pprint.pformat(influxDB_record))
245            self.write_api.write(self.bucket_id, self.org_id, influxDB_record)
246
247        except Exception as e:
248            if not self.quiet:
249                logging.warning('InfluxDBWriter exception: %s', str(e))
250                logging.warning('InfluxDBWriter could not ingest record '
251                                'type %s: %s', type(record), str(record))

Note: Assume record is a dict or DASRecord or list of dict/DASRecord. In each record look for 'fields', 'data_id' and 'timestamp' (UTC epoch seconds). If data_id is missing, use the bucket_name we were initialized with.