Update code and dependencies to support python3.10 to python3.11

Author: nzlosh

.github/workflows/check.yml

@@ -0,0 +1,31 @@
+name: Check-Build-Publish
+
+on:
+  push:
+    branches: ['*']
+  pull_request:
+    branches:
+      - master
+  schedule:
+    # run every Monday at 6am
+    - cron: '0 6 * * 1'
+
+jobs:
+  tests:
+    runs-on: ubuntu-22.04
+    strategy:
+      matrix:
+        python-version: ["3.10", "3.11", "3.12"]
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+      - name: "Set up Python ${{ matrix.python-version }}"
+        uses: actions/setup-python@v4
+        with:
+          python-version: "${{ matrix.python-version }}"
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip tox
+          tox
+      - name: "Run tox for ${{ matrix.python-version }}"
+        run: make check

logshipper/cmd.py

@@ -22,6 +22,7 @@
 import time
 
 import eventlet
+
 eventlet.monkey_patch()
 
 import logshipper.pipeline
@@ -32,29 +33,28 @@
 
 def main():
     global LOG, ARGS
-    parser = argparse.ArgumentParser(
-        description="Processes log messages and sends them elsewhere")
+    parser = argparse.ArgumentParser(description="Processes log messages and sends them elsewhere")
 
-    parser.add_argument('pipeline', nargs='+',
-                        help='Where to find pipelines (*.yml files)')
+    parser.add_argument("pipeline", nargs="+", help="Where to find pipelines (*.yml files)")
 
-    parser.add_argument('--verbose', action='store_true')
-    parser.add_argument('--debug', action='store_true')
+    parser.add_argument("--verbose", action="store_true")
+    parser.add_argument("--debug", action="store_true")
 
     ARGS = parser.parse_args()
 
     if ARGS.debug:
-        log_level = 'DEBUG'
+        log_level = "DEBUG"
     elif ARGS.verbose:
-        log_level = 'INFO'
+        log_level = "INFO"
     else:
-        log_level = 'WARNING'
+        log_level = "WARNING"
 
     logging.basicConfig(level=log_level)
     LOG = logging.getLogger(__name__)
 
     pipeline_manager = logshipper.pipeline.PipelineManager(
-        [os.path.abspath(p) for p in ARGS.pipeline])
+        [os.path.abspath(p) for p in ARGS.pipeline]
+    )
 
     pipeline_manager.start()
 
@@ -67,49 +67,50 @@ def main():
 
 def ship_file():
     global LOG, ARGS
-    parser = argparse.ArgumentParser(
-        description="Processes log messages and sends them elsewhere")
+    parser = argparse.ArgumentParser(description="Processes log messages and sends them elsewhere")
 
-    parser.add_argument('--pipeline', required=True)
+    parser.add_argument("--pipeline", required=True)
 
-    parser.add_argument('--pipeline-path', action='append',
-                        help='Where to find pipelines (*.yml files)')
+    parser.add_argument(
+        "--pipeline-path", action="append", help="Where to find pipelines (*.yml files)"
+    )
 
-    parser.add_argument('file', nargs='+', help='File to ship')
+    parser.add_argument("file", nargs="+", help="File to ship")
 
-    parser.add_argument('--verbose', action='store_true')
-    parser.add_argument('--debug', action='store_true')
+    parser.add_argument("--verbose", action="store_true")
+    parser.add_argument("--debug", action="store_true")
 
     ARGS = parser.parse_args()
 
     if ARGS.debug:
-        log_level = 'DEBUG'
+        log_level = "DEBUG"
     elif ARGS.verbose:
-        log_level = 'INFO'
+        log_level = "INFO"
     else:
-        log_level = 'WARNING'
+        log_level = "WARNING"
 
     logging.basicConfig(level=log_level)
     LOG = logging.getLogger(__name__)
 
     pipeline_manager = logshipper.pipeline.PipelineManager(
-        [os.path.abspath(p) for p in ARGS.pipeline_path])
+        [os.path.abspath(p) for p in ARGS.pipeline_path]
+    )
 
     pipeline_manager.load_pipelines()
 
     for filename in ARGS.file:
         # Peek to detect gzip
-        with open(filename, 'rb') as f:
+        with open(filename, "rb") as f:
             header = f.read(16)
 
         if header[0:2] == b"\037\213":
-            print("Processing gzipped file %s" % filename)
+            print(("Processing gzipped file %s" % filename))
             file_handle = gzip.open(filename)
-        elif header[0:3] == b"\x42\x5A\x68":
-            print("Processing bz2'ed file %s" % filename)
+        elif header[0:3] == b"\x42\x5a\x68":
+            print(("Processing bz2'ed file %s" % filename))
             file_handle = bz2.BZ2File(filename)
         else:
-            print("Processing uncompressed file %s" % filename)
+            print(("Processing uncompressed file %s" % filename))
             file_handle = open(filename)
 
         with file_handle:
@@ -120,8 +121,7 @@ def ship_file():
                 lines += 1
                 line = line.rstrip("\r\n")
                 try:
-                    pipeline_manager.process({'message': line},
-                                             ARGS.pipeline)
+                    pipeline_manager.process({"message": line}, ARGS.pipeline)
                 except KeyboardInterrupt:
                     break
                 except Exception:
@@ -131,8 +131,7 @@ def ship_file():
 
                 if (time.time() - treport) > 1:
                     rate = lines / (time.time() - start_time)
-                    sys.stdout.write(
-                        "Processing %i lines, %4.1f/s\r" % (lines, rate))
+                    sys.stdout.write("Processing %i lines, %4.1f/s\r" % (lines, rate))
                     treport = time.time()
                 sys.stdout.flush()
 

logshipper/context.py

@@ -20,9 +20,11 @@
 
 
 def _template_code(template):
-    if (template in (True, False, None) or
-            isinstance(template, six.integer_types) or
-            isinstance(template, float)):
+    if (
+        template in (True, False, None)
+        or isinstance(template, six.integer_types)
+        or isinstance(template, float)
+    ):
         return {}, repr(template)
 
     if isinstance(template, six.string_types):
@@ -44,7 +46,7 @@ def _template_code(template):
         result = ["{"]
         namespace = {}
 
-        for key, value in template.items():
+        for key, value in list(template.items()):
             sub_namespace, value_code = _template_code(value)
             result.append("%r:%s," % (key, value_code))
             namespace.update(sub_namespace)
@@ -83,44 +85,41 @@ def _template_code_string(template):
 
             if conversion:
                 namespace["fmt"] = fmt
-                value_code = "fmt.convert_field(%s, %r)" % (value_code,
-                                                            conversion)
+                value_code = "fmt.convert_field(%s, %r)" % (value_code, conversion)
 
             if not format_spec:
                 result.append("str(%s)" % value_code)
-            elif '{' in format_spec:
+            elif "{" in format_spec:
                 namespace["fmt"] = fmt
-                result.append("format(%s, fmt.vformat(%r, args, kwargs))" %
-                              (value_code, format_spec))
+                result.append(
+                    "format(%s, fmt.vformat(%r, args, kwargs))" % (value_code, format_spec)
+                )
             else:
                 result.append("format(%s, %r)\n" % (value_code, format_spec))
 
     if not result:
-        return namespace, "\"\""
+        return namespace, '""'
     if len(result) == 1:
         return namespace, result[0]
 
-    return namespace, "\"\".join([%s])" % (", ".join(result))
+    return namespace, '"".join([%s])' % (", ".join(result))
 
 
 def prepare_template(template):
     namespace, code = _template_code(template)
 
-    result = ("def template(args, kwargs):\n"
-              "  return %s" % code)
+    result = "def template(args, kwargs):\n" "  return %s" % code
 
     six.exec_(result, namespace)
 
     func = namespace["template"]
-    func.interpolate = lambda context: func(context.backreferences,
-                                            context.message)
+    func.interpolate = lambda context: func(context.backreferences, context.message)
 
     return func
 
 
 class Context(object):
-    __slots__ = ['pipeline_manager', 'message', 'match', 'match_field',
-                 'backreferences', 'matches']
+    __slots__ = ["pipeline_manager", "message", "match", "match_field", "backreferences", "matches"]
 
     def __init__(self, message, pipeline_manager):
         self.pipeline_manager = pipeline_manager

logshipper/elasticsearch.py

@@ -29,12 +29,12 @@ def json_default(value):
     else:
         return str(value)
 
+
 TRUE_VALUES = set([True, 1, "yes", "true", "on"])
 
 
 def to_bool(value):
-    return (value in TRUE_VALUES or
-            str(value).lower() in TRUE_VALUES)
+    return value in TRUE_VALUES or str(value).lower() in TRUE_VALUES
 
 
 def md5_hash(data):
@@ -65,31 +65,32 @@ def prepare_elasticsearch_http(parameters):
         ```http://localhost:9200/```.
     """
 
-    index = parameters.get('index', 'logshipper-{timestamp:%Y.%m.%d}')
+    index = parameters.get("index", "logshipper-{timestamp:%Y.%m.%d}")
     index = logshipper.context.prepare_template(index)
 
-    if 'id' in parameters:
-        id_ = logshipper.context.prepare_template(parameters['id']).interpolate
+    if "id" in parameters:
+        id_ = logshipper.context.prepare_template(parameters["id"]).interpolate
     else:
         id_ = None
 
-    doctype = parameters.get('doctype', 'log')
-    timestamp_field = parameters.get('timestamp', '@timestamp')
+    doctype = parameters.get("doctype", "log")
+    timestamp_field = parameters.get("timestamp", "@timestamp")
+
+    if "document" in parameters:
+        document_template = logshipper.context.prepare_template(parameters["document"]).interpolate
+    elif timestamp_field != "timestamp":
 
-    if 'document' in parameters:
-        document_template = logshipper.context.prepare_template(
-            parameters['document']).interpolate
-    elif timestamp_field != 'timestamp':
         def document_template(context):
             result = dict(context.message)
             result[timestamp_field] = result.pop("timestamp")
             return result
+
     else:
         document_template = lambda context: dict(context.message)
 
-    sort_keys = to_bool(parameters.get('sort_keys', False))
+    sort_keys = to_bool(parameters.get("sort_keys", False))
 
-    base_url = parameters.get('url', "http://localhost:9200/")
+    base_url = parameters.get("url", "http://localhost:9200/")
     if not base_url.endswith("/"):
         base_url += "/"
 
@@ -98,11 +99,14 @@ def document_template(context):
     def handle_elasticsearch_http(message, context):
         document = document_template(context)
 
-        document = json.dumps(document, default=json_default,
-                              sort_keys=sort_keys).encode('utf8')
+        document = json.dumps(document, default=json_default, sort_keys=sort_keys).encode("utf8")
 
-        url = "%s%s/%s/%s" % (base_url, index.interpolate(context), doctype,
-                              id_(context) if id_ else md5_hash(document))
+        url = "%s%s/%s/%s" % (
+            base_url,
+            index.interpolate(context),
+            doctype,
+            id_(context) if id_ else md5_hash(document),
+        )
 
         result = session.put(url, data=document)
         result.raise_for_status()