From 97d8c06b615c61bdb616bcd719c24a75149ae808 Mon Sep 17 00:00:00 2001 From: srv-rr-github-token <94607705+srv-rr-github-token@users.noreply.github.com> Date: Tue, 26 Nov 2024 10:40:38 +0000 Subject: [PATCH 1/7] chore(release): 5.5.0 # [5.5.0](https://github.com/splunk/addonfactory-solutions-library-python/compare/v5.4.0...v5.5.0) (2024-11-26) ### Features * add function to get the proxy details ([#406](https://github.com/splunk/addonfactory-solutions-library-python/issues/406)) ([c9e1f73](https://github.com/splunk/addonfactory-solutions-library-python/commit/c9e1f7350620963422747d90cfde529d3bb17511)) --- pyproject.toml | 2 +- solnlib/__init__.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index d9b580ef..673bdd52 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,7 @@ [tool.poetry] name = "solnlib" -version = "5.5.0-beta.1" +version = "5.5.0" description = "The Splunk Software Development Kit for Splunk Solutions" authors = ["Splunk "] license = "Apache-2.0" diff --git a/solnlib/__init__.py b/solnlib/__init__.py index 2f45611c..0d92fb1f 100644 --- a/solnlib/__init__.py +++ b/solnlib/__init__.py @@ -56,4 +56,4 @@ "utils", ] -__version__ = "5.5.0-beta.1" +__version__ = "5.5.0" From a6fa555ec8e724654a03fe200a94f643fd1c229e Mon Sep 17 00:00:00 2001 From: sgoral-splunk <138458044+sgoral-splunk@users.noreply.github.com> Date: Thu, 28 Nov 2024 16:03:03 +0100 Subject: [PATCH 2/7] feat: removing requests and urllib3 dependencies (#413) BREAKING CHANGE: removing requests and urllib3 dependencies --- docs/release_6_0_0.md | 16 ++++++++++++++++ mkdocs.yml | 1 + poetry.lock | 3 +-- pyproject.toml | 2 -- 4 files changed, 18 insertions(+), 4 deletions(-) create mode 100644 docs/release_6_0_0.md diff --git a/docs/release_6_0_0.md b/docs/release_6_0_0.md new file mode 100644 index 00000000..63b3af92 --- /dev/null +++ b/docs/release_6_0_0.md @@ -0,0 +1,16 @@ +# Removed requests and urllib3 from solnlib +The `requests` and `urllib3` libraries has been removed from solnlib, so solnlib now depends on the `requests` and `urllib3` libraries from the running environment. +By default, Splunk delivers the above libraries and their version depends on the Splunk version. More information [here](https://docs.splunk.com/Documentation/Splunk/9.2.3/ReleaseNotes/Credits). + +**IMPORTANT**: `urllib3` is available in Splunk `v8.1.0` and later + +Please note that if `requests` or `urllib3` are installed in `/lib` e.g. as a dependency of another library, that version will be taken first. +If `requests` or `urllib3` is missing in the add-on's `lib` directory, the version provided by Splunk will be used. + +## Custom Version of requests and urllib3 +In case the Splunk's `requests` or `urllib3` version is not sufficient for you, +you can deliver version you need by simply adding it to the `requirements.txt` or `pyproject.toml` file in your add-on. + +## Use solnlib outside the Splunk +**Solnlib** no longer provides `requests` and `urllib3` so if you want to use **solnlib** outside the Splunk, please note that you will need to +provide these libraries yourself in the environment where **solnlib** is used. diff --git a/mkdocs.yml b/mkdocs.yml index 5aa2bfaa..f0742b0e 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -30,6 +30,7 @@ plugins: nav: - Home: index.md + - Release 6.0.0: release_6_0_0.md - References: - modular_input: - "checkpointer.py": modular_input/checkpointer.md diff --git a/poetry.lock b/poetry.lock index 49a816f5..022d5f55 100644 --- a/poetry.lock +++ b/poetry.lock @@ -727,7 +727,6 @@ files = [ {file = "PyYAML-6.0.1-cp311-cp311-win_amd64.whl", hash = "sha256:bf07ee2fef7014951eeb99f56f39c9bb4af143d8aa3c21b1677805985307da34"}, {file = "PyYAML-6.0.1-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:855fb52b0dc35af121542a76b9a84f8d1cd886ea97c84703eaa6d88e37a2ad28"}, {file = "PyYAML-6.0.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:40df9b996c2b73138957fe23a16a4f0ba614f4c0efce1e9406a184b6d07fa3a9"}, - {file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a08c6f0fe150303c1c6b71ebcd7213c2858041a7e01975da3a99aed1e7a378ef"}, {file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6c22bec3fbe2524cde73d7ada88f6566758a8f7227bfbf93a408a9d86bcc12a0"}, {file = "PyYAML-6.0.1-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:8d4e9c88387b0f5c7d5f281e55304de64cf7f9c0021a3525bd3b1c542da3b0e4"}, {file = "PyYAML-6.0.1-cp312-cp312-win32.whl", hash = "sha256:d483d2cdf104e7c9fa60c544d92981f12ad66a457afae824d146093b8c294c54"}, @@ -1024,4 +1023,4 @@ testing = ["big-O", "flake8 (<5)", "jaraco.functools", "jaraco.itertools", "more [metadata] lock-version = "2.0" python-versions = ">=3.7,<3.14" -content-hash = "355e3165623b332a6856b7959863e7dbf61f9f22fc0b880aede1d84cc3c53dba" +content-hash = "e24950c3694801361bf9956c190d77a0c5f2fdb82786f4a101ab50de34932920" diff --git a/pyproject.toml b/pyproject.toml index 673bdd52..d1ffef0e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,8 +40,6 @@ classifiers = [ [tool.poetry.dependencies] python = ">=3.7,<3.14" -requests = "^2.31.0" -urllib3 = "<2" sortedcontainers = ">=2" defusedxml = ">=0.7" splunk-sdk = ">=1.6" From ce6a2ca243f8aa0e3165035bd97551de4ef3bf5b Mon Sep 17 00:00:00 2001 From: srv-rr-github-token <94607705+srv-rr-github-token@users.noreply.github.com> Date: Thu, 28 Nov 2024 15:08:21 +0000 Subject: [PATCH 3/7] chore(release): 6.0.0 # [6.0.0](https://github.com/splunk/addonfactory-solutions-library-python/compare/v5.5.0...v6.0.0) (2024-11-28) ### Features * removing requests and urllib3 dependencies ([#413](https://github.com/splunk/addonfactory-solutions-library-python/issues/413)) ([a6fa555](https://github.com/splunk/addonfactory-solutions-library-python/commit/a6fa555ec8e724654a03fe200a94f643fd1c229e)) ### BREAKING CHANGES * removing requests and urllib3 dependencies --- pyproject.toml | 2 +- solnlib/__init__.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index d1ffef0e..b6d5357c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,7 @@ [tool.poetry] name = "solnlib" -version = "5.5.0" +version = "6.0.0" description = "The Splunk Software Development Kit for Splunk Solutions" authors = ["Splunk "] license = "Apache-2.0" diff --git a/solnlib/__init__.py b/solnlib/__init__.py index 0d92fb1f..136a4b9f 100644 --- a/solnlib/__init__.py +++ b/solnlib/__init__.py @@ -56,4 +56,4 @@ "utils", ] -__version__ = "5.5.0" +__version__ = "6.0.0" From f50bf52abdc7ce1afb8f50f8a427e6c70421c3ac Mon Sep 17 00:00:00 2001 From: Artem Rys Date: Fri, 29 Nov 2024 13:07:46 +0100 Subject: [PATCH 4/7] fix: require splunk-sdk at least 2.0.2 (#414) Jira: https://splunk.atlassian.net/browse/ADDON-77054 This PR makes sure that `solnlib` will require at least `2.0.2` version of `splunk-sdk`. --- poetry.lock | 3 ++- pyproject.toml | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/poetry.lock b/poetry.lock index 022d5f55..c5661738 100644 --- a/poetry.lock +++ b/poetry.lock @@ -727,6 +727,7 @@ files = [ {file = "PyYAML-6.0.1-cp311-cp311-win_amd64.whl", hash = "sha256:bf07ee2fef7014951eeb99f56f39c9bb4af143d8aa3c21b1677805985307da34"}, {file = "PyYAML-6.0.1-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:855fb52b0dc35af121542a76b9a84f8d1cd886ea97c84703eaa6d88e37a2ad28"}, {file = "PyYAML-6.0.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:40df9b996c2b73138957fe23a16a4f0ba614f4c0efce1e9406a184b6d07fa3a9"}, + {file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a08c6f0fe150303c1c6b71ebcd7213c2858041a7e01975da3a99aed1e7a378ef"}, {file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6c22bec3fbe2524cde73d7ada88f6566758a8f7227bfbf93a408a9d86bcc12a0"}, {file = "PyYAML-6.0.1-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:8d4e9c88387b0f5c7d5f281e55304de64cf7f9c0021a3525bd3b1c542da3b0e4"}, {file = "PyYAML-6.0.1-cp312-cp312-win32.whl", hash = "sha256:d483d2cdf104e7c9fa60c544d92981f12ad66a457afae824d146093b8c294c54"}, @@ -1023,4 +1024,4 @@ testing = ["big-O", "flake8 (<5)", "jaraco.functools", "jaraco.itertools", "more [metadata] lock-version = "2.0" python-versions = ">=3.7,<3.14" -content-hash = "e24950c3694801361bf9956c190d77a0c5f2fdb82786f4a101ab50de34932920" +content-hash = "7296af4efc8a409147dfcfc840895f32432578649836d5906506225d3be8e58a" diff --git a/pyproject.toml b/pyproject.toml index b6d5357c..23c1a75d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -42,7 +42,7 @@ classifiers = [ python = ">=3.7,<3.14" sortedcontainers = ">=2" defusedxml = ">=0.7" -splunk-sdk = ">=1.6" +splunk-sdk = ">=2.0.2" [tool.poetry.group.dev.dependencies] pytest = ">=7" From d46cd2194272a2db1493251997b7623813613570 Mon Sep 17 00:00:00 2001 From: srv-rr-github-token <94607705+srv-rr-github-token@users.noreply.github.com> Date: Fri, 29 Nov 2024 12:13:30 +0000 Subject: [PATCH 5/7] chore(release): 6.0.1 ## [6.0.1](https://github.com/splunk/addonfactory-solutions-library-python/compare/v6.0.0...v6.0.1) (2024-11-29) ### Bug Fixes * require splunk-sdk at least 2.0.2 ([#414](https://github.com/splunk/addonfactory-solutions-library-python/issues/414)) ([f50bf52](https://github.com/splunk/addonfactory-solutions-library-python/commit/f50bf52abdc7ce1afb8f50f8a427e6c70421c3ac)) --- pyproject.toml | 2 +- solnlib/__init__.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 23c1a75d..f85ee9e3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,7 @@ [tool.poetry] name = "solnlib" -version = "6.0.0" +version = "6.0.1" description = "The Splunk Software Development Kit for Splunk Solutions" authors = ["Splunk "] license = "Apache-2.0" diff --git a/solnlib/__init__.py b/solnlib/__init__.py index 136a4b9f..245e4a96 100644 --- a/solnlib/__init__.py +++ b/solnlib/__init__.py @@ -56,4 +56,4 @@ "utils", ] -__version__ = "6.0.0" +__version__ = "6.0.1" From 4953486baf202f7359fedbe29c1f0d95232e4caf Mon Sep 17 00:00:00 2001 From: Hetang Modi <62056057+hetangmodi-crest@users.noreply.github.com> Date: Mon, 2 Dec 2024 20:16:34 +0530 Subject: [PATCH 6/7] feat: moved methods from splunktalib (#415) [ADDON-73693](https://splunk.atlassian.net/browse/ADDON-73693) Migrated all the methods of `splunktalib` which were used in `splunktaucclib` to `solnlib` --- solnlib/concurrent/concurrent_executor.py | 102 +++++++ solnlib/concurrent/process_pool.py | 75 +++++ solnlib/concurrent/thread_pool.py | 347 ++++++++++++++++++++++ solnlib/modular_input/modinput.py | 161 ++++++++++ solnlib/rest.py | 95 ++++++ solnlib/schedule/job.py | 122 ++++++++ solnlib/schedule/scheduler.py | 162 ++++++++++ solnlib/utils.py | 26 ++ 8 files changed, 1090 insertions(+) create mode 100644 solnlib/concurrent/concurrent_executor.py create mode 100644 solnlib/concurrent/process_pool.py create mode 100644 solnlib/concurrent/thread_pool.py create mode 100644 solnlib/modular_input/modinput.py create mode 100644 solnlib/rest.py create mode 100644 solnlib/schedule/job.py create mode 100644 solnlib/schedule/scheduler.py diff --git a/solnlib/concurrent/concurrent_executor.py b/solnlib/concurrent/concurrent_executor.py new file mode 100644 index 00000000..6d3c7521 --- /dev/null +++ b/solnlib/concurrent/concurrent_executor.py @@ -0,0 +1,102 @@ +# +# Copyright 2024 Splunk Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Concurrent executor provides concurrent executing function either in a +thread pool or a process pool.""" + +import solnlib.concurrent.process_pool as pp +import solnlib.concurrent.thread_pool as tp + + +class ConcurrentExecutor: + def __init__(self, config): + """ + :param config: dict like object, contains thread_min_size (int), + thread_max_size (int), daemonize_thread (bool), + process_size (int) + """ + + self._io_executor = tp.ThreadPool( + config.get("thread_min_size", 0), + config.get("thread_max_size", 0), + config.get("task_queue_size", 1024), + config.get("daemonize_thread", True), + ) + self._compute_executor = None + if config.get("process_size", 0): + self._compute_executor = pp.ProcessPool(config.get("process_size", 0)) + + def start(self): + self._io_executor.start() + + def tear_down(self): + self._io_executor.tear_down() + if self._compute_executor is not None: + self._compute_executor.tear_down() + + def run_io_func_sync(self, func, args=(), kwargs=None): + """ + :param func: callable + :param args: free params + :param kwargs: named params + :return whatever the func returns + """ + + return self._io_executor.apply(func, args, kwargs) + + def run_io_func_async(self, func, args=(), kwargs=None, callback=None): + """ + :param func: callable + :param args: free params + :param kwargs: named params + :calllback: when func is done and without exception, call the callback + :return whatever the func returns + """ + + return self._io_executor.apply_async(func, args, kwargs, callback) + + def enqueue_io_funcs(self, funcs, block=True): + """run jobs in a fire and forget way, no result will be handled over to + clients. + + :param funcs: tuple/list-like or generator like object, func shall be + callable + """ + + return self._io_executor.enqueue_funcs(funcs, block) + + def run_compute_func_sync(self, func, args=(), kwargs={}): + """ + :param func: callable + :param args: free params + :param kwargs: named params + :return whatever the func returns + """ + + assert self._compute_executor is not None + return self._compute_executor.apply(func, args, kwargs) + + def run_compute_func_async(self, func, args=(), kwargs={}, callback=None): + """ + :param func: callable + :param args: free params + :param kwargs: named params + :calllback: when func is done and without exception, call the callback + :return whatever the func returns + """ + + assert self._compute_executor is not None + return self._compute_executor.apply_async(func, args, kwargs, callback) diff --git a/solnlib/concurrent/process_pool.py b/solnlib/concurrent/process_pool.py new file mode 100644 index 00000000..723d98e0 --- /dev/null +++ b/solnlib/concurrent/process_pool.py @@ -0,0 +1,75 @@ +# +# Copyright 2024 Splunk Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""A wrapper of multiprocessing.pool.""" + +import multiprocessing + +import logging + + +class ProcessPool: + """A simple wrapper of multiprocessing.pool.""" + + def __init__(self, size=0, maxtasksperchild=10000): + if size <= 0: + size = multiprocessing.cpu_count() + self.size = size + self._pool = multiprocessing.Pool( + processes=size, maxtasksperchild=maxtasksperchild + ) + self._stopped = False + + def tear_down(self): + """Tear down the pool.""" + + if self._stopped: + logging.info("ProcessPool has already stopped.") + return + self._stopped = True + + self._pool.close() + self._pool.join() + logging.info("ProcessPool stopped.") + + def apply(self, func, args=(), kwargs={}): + """ + :param func: callable + :param args: free params + :param kwargs: named params + :return whatever the func returns + """ + + if self._stopped: + logging.info("ProcessPool has already stopped.") + return None + + return self._pool.apply(func, args, kwargs) + + def apply_async(self, func, args=(), kwargs={}, callback=None): + """ + :param func: callable + :param args: free params + :param kwargs: named params + :callback: when func is done without exception, call this callack + :return whatever the func returns + """ + + if self._stopped: + logging.info("ProcessPool has already stopped.") + return None + + return self._pool.apply_async(func, args, kwargs, callback) diff --git a/solnlib/concurrent/thread_pool.py b/solnlib/concurrent/thread_pool.py new file mode 100644 index 00000000..2d0b5bcd --- /dev/null +++ b/solnlib/concurrent/thread_pool.py @@ -0,0 +1,347 @@ +# +# Copyright 2024 Splunk Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""A simple thread pool implementation.""" + +import multiprocessing +import queue +import threading +import traceback +from time import time +import logging + + +class ThreadPool: + """A simple thread pool implementation.""" + + _high_watermark = 0.2 + _resize_window = 10 + + def __init__(self, min_size=1, max_size=128, task_queue_size=1024, daemon=True): + assert task_queue_size + + if not min_size or min_size <= 0: + min_size = multiprocessing.cpu_count() + + if not max_size or max_size <= 0: + max_size = multiprocessing.cpu_count() * 8 + + self._min_size = min_size + self._max_size = max_size + self._daemon = daemon + + self._work_queue = queue.Queue(task_queue_size) + self._thrs = [] + for _ in range(min_size): + thr = threading.Thread(target=self._run) + self._thrs.append(thr) + self._admin_queue = queue.Queue() + self._admin_thr = threading.Thread(target=self._do_admin) + self._last_resize_time = time() + self._last_size = min_size + self._lock = threading.Lock() + self._occupied_threads = 0 + self._count_lock = threading.Lock() + self._started = False + + def start(self): + """Start threads in the pool.""" + + with self._lock: + if self._started: + return + self._started = True + + for thr in self._thrs: + thr.daemon = self._daemon + thr.start() + + self._admin_thr.start() + logging.info("ThreadPool started.") + + def tear_down(self): + """Tear down thread pool.""" + + with self._lock: + if not self._started: + return + self._started = False + + for thr in self._thrs: + self._work_queue.put(None, block=True) + + self._admin_queue.put(None) + + if not self._daemon: + logging.info("Wait for threads to stop.") + for thr in self._thrs: + thr.join() + self._admin_thr.join() + + logging.info("ThreadPool stopped.") + + def enqueue_funcs(self, funcs, block=True): + """run jobs in a fire and forget way, no result will be handled over to + clients. + + :param funcs: tuple/list-like or generator like object, func shall be + callable + """ + + if not self._started: + logging.info("ThreadPool has already stopped.") + return + + for func in funcs: + self._work_queue.put(func, block) + + def apply_async(self, func, args=(), kwargs=None, callback=None): + """ + :param func: callable + :param args: free params + :param kwargs: named params + :callback: when func is done and without exception, call the callback + :return AsyncResult, clients can poll or wait the result through it + """ + + if not self._started: + logging.info("ThreadPool has already stopped.") + return None + + res = AsyncResult(func, args, kwargs, callback) + self._work_queue.put(res) + return res + + def apply(self, func, args=(), kwargs=None): + """ + :param func: callable + :param args: free params + :param kwargs: named params + :return whatever the func returns + """ + + if not self._started: + logging.info("ThreadPool has already stopped.") + return None + + res = self.apply_async(func, args, kwargs) + return res.get() + + def size(self): + return self._last_size + + def resize(self, new_size): + """Resize the pool size, spawn or destroy threads if necessary.""" + + if new_size <= 0: + return + + if self._lock.locked() or not self._started: + logging.info( + "Try to resize thread pool during the tear " "down process, do nothing" + ) + return + + with self._lock: + self._remove_exited_threads_with_lock() + size = self._last_size + self._last_size = new_size + if new_size > size: + for _ in range(new_size - size): + thr = threading.Thread(target=self._run) + thr.daemon = self._daemon + thr.start() + self._thrs.append(thr) + elif new_size < size: + for _ in range(size - new_size): + self._work_queue.put(None) + logging.info("Finished ThreadPool resizing. New size=%d", new_size) + + def _remove_exited_threads_with_lock(self): + """Join the exited threads last time when resize was called.""" + + joined_thrs = set() + for thr in self._thrs: + if not thr.is_alive(): + try: + if not thr.daemon: + thr.join(timeout=0.5) + joined_thrs.add(thr.ident) + except RuntimeError: + pass + + if joined_thrs: + live_thrs = [] + for thr in self._thrs: + if thr.ident not in joined_thrs: + live_thrs.append(thr) + self._thrs = live_thrs + + def _do_resize_according_to_loads(self): + if ( + self._last_resize_time + and time() - self._last_resize_time < self._resize_window + ): + return + + thr_size = self._last_size + free_thrs = thr_size - self._occupied_threads + work_size = self._work_queue.qsize() + + logging.debug( + "current_thr_size=%s, free_thrs=%s, work_size=%s", + thr_size, + free_thrs, + work_size, + ) + if work_size and work_size > free_thrs: + if thr_size < self._max_size: + thr_size = min(thr_size * 2, self._max_size) + self.resize(thr_size) + elif free_thrs > 0: + free = free_thrs * 1.0 + if free / thr_size >= self._high_watermark and free_thrs >= 2: + # 20 % thrs are idle, tear down half of the idle ones + thr_size = thr_size - int(free_thrs // 2) + if thr_size > self._min_size: + self.resize(thr_size) + self._last_resize_time = time() + + def _do_admin(self): + admin_q = self._admin_queue + resize_win = self._resize_window + while 1: + try: + wakup = admin_q.get(timeout=resize_win + 1) + except queue.Empty: + self._do_resize_according_to_loads() + continue + + if wakup is None: + break + else: + self._do_resize_according_to_loads() + logging.info( + "ThreadPool admin thread=%s stopped.", threading.current_thread().getName() + ) + + def _run(self): + """Threads callback func, run forever to handle jobs from the job + queue.""" + + work_queue = self._work_queue + count_lock = self._count_lock + while 1: + logging.debug("Going to get job") + func = work_queue.get() + if func is None: + break + + if not self._started: + break + + logging.debug("Going to exec job") + with count_lock: + self._occupied_threads += 1 + + try: + func() + except Exception: + logging.error(traceback.format_exc()) + + with count_lock: + self._occupied_threads -= 1 + + logging.debug("Done with exec job") + logging.info("Thread work_queue_size=%d", work_queue.qsize()) + + logging.debug("Worker thread %s stopped.", threading.current_thread().getName()) + + +class AsyncResult: + def __init__(self, func, args, kwargs, callback): + self._func = func + self._args = args + self._kwargs = kwargs + self._callback = callback + self._q = queue.Queue() + + def __call__(self): + try: + if self._args and self._kwargs: + res = self._func(*self._args, **self._kwargs) + elif self._args: + res = self._func(*self._args) + elif self._kwargs: + res = self._func(**self._kwargs) + else: + res = self._func() + except Exception as e: + self._q.put(e) + return + else: + self._q.put(res) + + if self._callback is not None: + self._callback() + + def get(self, timeout=None): + """Return the result when it arrives. + + If timeout is not None and the result does not arrive within + timeout seconds then multiprocessing.TimeoutError is raised. If + the remote call raised an exception then that exception will be + reraised by get(). + """ + + try: + res = self._q.get(timeout=timeout) + except queue.Empty: + raise multiprocessing.TimeoutError("Timed out") + + if isinstance(res, Exception): + raise res + return res + + def wait(self, timeout=None): + """Wait until the result is available or until timeout seconds pass.""" + + try: + res = self._q.get(timeout=timeout) + except queue.Empty: + pass + else: + self._q.put(res) + + def ready(self): + """Return whether the call has completed.""" + + return len(self._q) + + def successful(self): + """Return whether the call completed without raising an exception. + + Will raise AssertionError if the result is not ready. + """ + + if not self.ready(): + raise AssertionError("Function is not ready") + res = self._q.get() + self._q.put(res) + + if isinstance(res, Exception): + return False + return True diff --git a/solnlib/modular_input/modinput.py b/solnlib/modular_input/modinput.py new file mode 100644 index 00000000..1e6b4e4d --- /dev/null +++ b/solnlib/modular_input/modinput.py @@ -0,0 +1,161 @@ +# +# Copyright 2024 Splunk Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import subprocess +import sys +import traceback + +import solnlib.splunkenv as sp +import logging + + +def _parse_modinput_configs(root, outer_block, inner_block): + """When user splunkd spawns modinput script to do config check or run. + + + + localhost.localdomain + https://127.0.0.1:8089 + xxxyyyzzz + ckpt_dir + + + 60 + localhost.localdomain + snow + 10 + + ... + + + + When user create an stanza through data input on WebUI + + + + localhost.localdomain + https://127.0.0.1:8089 + xxxyyyzzz + ckpt_dir + + 60 + + localhost.localdomain + snow + 10 + + + """ + + confs = root.getElementsByTagName(outer_block) + if not confs: + logging.error("Invalid config, missing %s section", outer_block) + raise Exception(f"Invalid config, missing {outer_block} section") + + configs = [] + stanzas = confs[0].getElementsByTagName(inner_block) + for stanza in stanzas: + config = {} + stanza_name = stanza.getAttribute("name") + if not stanza_name: + logging.error("Invalid config, missing name") + raise Exception("Invalid config, missing name") + + config["name"] = stanza_name + params = stanza.getElementsByTagName("param") + for param in params: + name = param.getAttribute("name") + if ( + name + and param.firstChild + and param.firstChild.nodeType == param.firstChild.TEXT_NODE + ): + config[name] = param.firstChild.data + configs.append(config) + return configs + + +def parse_modinput_configs(config_str): + """ + @config_str: modinput XML configuration feed by splunkd + @return: meta_config and stanza_config + """ + + import defusedxml.minidom as xdm + + meta_configs = { + "server_host": None, + "server_uri": None, + "session_key": None, + "checkpoint_dir": None, + } + root = xdm.parseString(config_str) + doc = root.documentElement + for tag in meta_configs.keys(): + nodes = doc.getElementsByTagName(tag) + if not nodes: + logging.error("Invalid config, missing %s section", tag) + raise Exception("Invalid config, missing %s section", tag) + + if nodes[0].firstChild and nodes[0].firstChild.nodeType == nodes[0].TEXT_NODE: + meta_configs[tag] = nodes[0].firstChild.data + else: + logging.error("Invalid config, expect text ndoe") + raise Exception("Invalid config, expect text ndoe") + + if doc.nodeName == "input": + configs = _parse_modinput_configs(doc, "configuration", "stanza") + else: + configs = _parse_modinput_configs(root, "items", "item") + return meta_configs, configs + + +def get_modinput_configs_from_cli(modinput, modinput_stanza=None): + """ + @modinput: modinput name + @modinput_stanza: modinput stanza name, for multiple instance only + """ + + assert modinput + + splunkbin = sp.get_splunk_bin() + cli = [splunkbin, "cmd", "splunkd", "print-modinput-config", modinput] + if modinput_stanza: + cli.append(modinput_stanza) + + out, err = subprocess.Popen( + cli, stdout=subprocess.PIPE, stderr=subprocess.PIPE + ).communicate() + if err: + logging.error("Failed to get modinput configs with error: %s", err) + return None, None + else: + return parse_modinput_configs(out) + + +def get_modinput_config_str_from_stdin(): + """Get modinput from stdin which is feed by splunkd.""" + + try: + return sys.stdin.read(5000) + except Exception: + logging.error(traceback.format_exc()) + raise + + +def get_modinput_configs_from_stdin(): + config_str = get_modinput_config_str_from_stdin() + return parse_modinput_configs(config_str) diff --git a/solnlib/rest.py b/solnlib/rest.py new file mode 100644 index 00000000..4ff3abe1 --- /dev/null +++ b/solnlib/rest.py @@ -0,0 +1,95 @@ +# +# Copyright 2024 Splunk Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import json +import urllib.parse +from traceback import format_exc +from typing import Optional + +import requests + +import logging + + +def splunkd_request( + splunkd_uri, + session_key, + method="GET", + headers=None, + data=None, + timeout=300, + retry=1, + verify=False, +) -> Optional[requests.Response]: + + headers = headers if headers is not None else {} + headers["Authorization"] = f"Splunk {session_key}" + content_type = headers.get("Content-Type") + if not content_type: + content_type = headers.get("content-type") + + if not content_type: + content_type = "application/x-www-form-urlencoded" + headers["Content-Type"] = content_type + + if data is not None: + if content_type == "application/json": + data = json.dumps(data) + else: + data = urllib.parse.urlencode(data) + + msg_temp = "Failed to send rest request=%s, errcode=%s, reason=%s" + resp = None + for _ in range(retry): + try: + resp = requests.request( + method=method, + url=splunkd_uri, + data=data, + headers=headers, + timeout=timeout, + verify=verify, + ) + except Exception: + logging.error(msg_temp, splunkd_uri, "unknown", format_exc()) + else: + if resp.status_code not in (200, 201): + if not (method == "GET" and resp.status_code == 404): + logging.debug( + msg_temp, splunkd_uri, resp.status_code, code_to_msg(resp) + ) + else: + return resp + else: + return resp + + +def code_to_msg(response: requests.Response): + code_msg_tbl = { + 400: f"Request error. reason={response.text}", + 401: "Authentication failure, invalid access credentials.", + 402: "In-use license disables this feature.", + 403: "Insufficient permission.", + 404: "Requested endpoint does not exist.", + 409: f"Invalid operation for this endpoint. reason={response.text}", + 500: f"Unspecified internal server error. reason={response.text}", + 503: ( + "Feature is disabled in the configuration file. " + "reason={}".format(response.text) + ), + } + + return code_msg_tbl.get(response.status_code, response.text) diff --git a/solnlib/schedule/job.py b/solnlib/schedule/job.py new file mode 100644 index 00000000..0e2e5fc7 --- /dev/null +++ b/solnlib/schedule/job.py @@ -0,0 +1,122 @@ +# +# Copyright 2024 Splunk Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import threading +import time + + +class Job: + """Timer wraps the callback and timestamp related stuff.""" + + _ident = 0 + _lock = threading.Lock() + + def __init__(self, func, job_props, interval, when=None, job_id=None): + """ + @job_props: dict like object + @func: execution function + @interval: execution interval + @when: seconds from epoch + @job_id: a unique id for the job + """ + + self._props = job_props + self._func = func + if when is None: + self._when = time.time() + else: + self._when = when + self._interval = interval + + if job_id is not None: + self._id = job_id + else: + with Job._lock: + self._id = Job._ident + 1 + Job._ident = Job._ident + 1 + self._stopped = False + + def ident(self): + return self._id + + def get_interval(self): + return self._interval + + def set_interval(self, interval): + self._interval = interval + + def get_expiration(self): + return self._when + + def set_initial_due_time(self, when): + if self._when is None: + self._when = when + + def update_expiration(self): + self._when += self._interval + + def get(self, key, default): + return self._props.get(key, default) + + def get_props(self): + return self._props + + def set_props(self, props): + self._props = props + + def __cmp__(self, other): + if other is None: + return 1 + + self_k = (self.get_expiration(), self.ident()) + other_k = (other.get_expiration(), other.ident()) + + if self_k == other_k: + return 0 + elif self_k < other_k: + return -1 + else: + return 1 + + def __eq__(self, other): + return isinstance(other, Job) and (self.ident() == other.ident()) + + def __lt__(self, other): + return self.__cmp__(other) == -1 + + def __gt__(self, other): + return self.__cmp__(other) == 1 + + def __ne__(self, other): + return not self.__eq__(other) + + def __le__(self, other): + return self.__lt__(other) or self.__eq__(other) + + def __ge__(self, other): + return self.__gt__(other) or self.__eq__(other) + + def __hash__(self): + return self.ident() + + def __call__(self): + self._func(self) + + def stop(self): + self._stopped = True + + def stopped(self): + return self._stopped diff --git a/solnlib/schedule/scheduler.py b/solnlib/schedule/scheduler.py new file mode 100644 index 00000000..fb1de53f --- /dev/null +++ b/solnlib/schedule/scheduler.py @@ -0,0 +1,162 @@ +# +# Copyright 2024 Splunk Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import queue +import random +import threading +from time import time + +import logging + + +class Scheduler: + """A simple scheduler which schedules the periodic or once event.""" + + import sortedcontainers as sc + + max_delay_time = 60 + + def __init__(self): + self._jobs = Scheduler.sc.SortedSet() + self._wakeup_q = queue.Queue() + self._lock = threading.Lock() + self._thr = threading.Thread(target=self._do_jobs) + # FIXME: the `daemon` property HAS to be passed in init() call ONLY, + # the below attribute setting is of incorrect spelling + self._thr.deamon = True + self._started = False + + def start(self): + """Start the schduler which will start the internal thread for + scheduling jobs. + + Please do tear_down when doing cleanup + """ + + if self._started: + logging.info("Scheduler already started.") + return + self._started = True + + self._thr.start() + + def tear_down(self): + """Stop the schduler which will stop the internal thread for scheduling + jobs.""" + + if not self._started: + logging.info("Scheduler already tear down.") + return + + self._wakeup_q.put(True) + + def _do_jobs(self): + while 1: + (sleep_time, jobs) = self.get_ready_jobs() + self._do_execution(jobs) + try: + done = self._wakeup_q.get(timeout=sleep_time) + except queue.Empty: + pass + else: + if done: + break + self._started = False + logging.info("Scheduler exited.") + + def get_ready_jobs(self): + """ + @return: a 2 element tuple. The first element is the next ready + duration. The second element is ready jobs list + """ + + now = time() + ready_jobs = [] + sleep_time = 1 + + with self._lock: + job_set = self._jobs + total_jobs = len(job_set) + for job in job_set: + if job.get_expiration() <= now: + ready_jobs.append(job) + + if ready_jobs: + del job_set[: len(ready_jobs)] + + for job in ready_jobs: + if job.get_interval() != 0 and not job.stopped(): + # repeated job, calculate next due time and enqueue + job.update_expiration() + job_set.add(job) + + if job_set: + sleep_time = job_set[0].get_expiration() - now + if sleep_time < 0: + logging.warn("Scheduler satuation, sleep_time=%s", sleep_time) + sleep_time = 0.1 + + if ready_jobs: + logging.info( + "Get %d ready jobs, next duration is %f, " + "and there are %s jobs scheduling", + len(ready_jobs), + sleep_time, + total_jobs, + ) + + ready_jobs.sort(key=lambda job: job.get("priority", 0), reverse=True) + return (sleep_time, ready_jobs) + + def add_jobs(self, jobs): + with self._lock: + now = time() + job_set = self._jobs + for job in jobs: + delay_time = random.randrange(0, self.max_delay_time) + job.set_initial_due_time(now + delay_time) + job_set.add(job) + self._wakeup() + + def update_jobs(self, jobs): + with self._lock: + job_set = self._jobs + for njob in jobs: + job_set.discard(njob) + job_set.add(njob) + self._wakeup() + + def remove_jobs(self, jobs): + with self._lock: + job_set = self._jobs + for njob in jobs: + njob.stop() + job_set.discard(njob) + self._wakeup() + + def number_of_jobs(self): + with self._lock: + return len(self._jobs) + + def disable_randomization(self): + self.max_delay_time = 1 + + def _wakeup(self): + self._wakeup_q.put(None) + + def _do_execution(self, jobs): + for job in jobs: + job() diff --git a/solnlib/utils.py b/solnlib/utils.py index 6aaa8777..4ff4fdc0 100644 --- a/solnlib/utils.py +++ b/solnlib/utils.py @@ -193,3 +193,29 @@ def extract_http_scheme_host_port(http_url: str) -> Tuple: if not http_info.scheme or not http_info.hostname or not http_info.port: raise ValueError(http_url + " is not in http(s)://hostname:port format") return http_info.scheme, http_info.hostname, http_info.port + + +def get_appname_from_path(absolute_path): + """Gets name of the app from its path. + + Arguments: + absolute_path: path of app + + Returns: + """ + absolute_path = os.path.normpath(absolute_path) + parts = absolute_path.split(os.path.sep) + parts.reverse() + for key in ("apps", "peer-apps", "manager-apps"): + try: + idx = parts.index(key) + except ValueError: + continue + else: + try: + if parts[idx + 1] == "etc": + return parts[idx - 1] + except IndexError: + pass + continue + return "-" From a0d0e03b527344c7702dde338a74e76cce99f91f Mon Sep 17 00:00:00 2001 From: srv-rr-github-token <94607705+srv-rr-github-token@users.noreply.github.com> Date: Mon, 2 Dec 2024 15:17:49 +0000 Subject: [PATCH 7/7] chore(release): 6.1.0-beta.1 # [6.1.0-beta.1](https://github.com/splunk/addonfactory-solutions-library-python/compare/v6.0.1...v6.1.0-beta.1) (2024-12-02) ### Features * moved methods from splunktalib ([#415](https://github.com/splunk/addonfactory-solutions-library-python/issues/415)) ([4953486](https://github.com/splunk/addonfactory-solutions-library-python/commit/4953486baf202f7359fedbe29c1f0d95232e4caf)) --- pyproject.toml | 2 +- solnlib/__init__.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index f85ee9e3..cb50faf1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,7 @@ [tool.poetry] name = "solnlib" -version = "6.0.1" +version = "6.1.0-beta.1" description = "The Splunk Software Development Kit for Splunk Solutions" authors = ["Splunk "] license = "Apache-2.0" diff --git a/solnlib/__init__.py b/solnlib/__init__.py index 245e4a96..01af8fed 100644 --- a/solnlib/__init__.py +++ b/solnlib/__init__.py @@ -56,4 +56,4 @@ "utils", ] -__version__ = "6.0.1" +__version__ = "6.1.0-beta.1"