Skip to content

checkpointer.py

This module provides two kinds of checkpointer: KVStoreCheckpointer, FileCheckpointer for modular input to save checkpoint.

__all__ = ['CheckpointerException', 'KVStoreCheckpointer', 'FileCheckpointer'] module-attribute

Checkpointer

Base class of checkpointer.

Source code in solnlib/modular_input/checkpointer.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class Checkpointer(metaclass=ABCMeta):
    """Base class of checkpointer."""

    @abstractmethod
    def update(self, key: str, state: Any):
        """Updates document with an id that equals to `key` and `state` as
        document data."""

    @abstractmethod
    def batch_update(self, states: Iterable[Dict[str, Any]]):
        """Updates multiple documents."""

    @abstractmethod
    def get(self, key: str) -> dict:
        """Gets document with an id that equals to `key`."""

    @abstractmethod
    def delete(self, key: str):
        """Deletes document with an id that equals to `key`."""

batch_update(states) abstractmethod

Updates multiple documents.

Source code in solnlib/modular_input/checkpointer.py 
49
50
51
@abstractmethod
def batch_update(self, states: Iterable[Dict[str, Any]]):
    """Updates multiple documents."""

delete(key) abstractmethod

Deletes document with an id that equals to key.

Source code in solnlib/modular_input/checkpointer.py 
57
58
59
@abstractmethod
def delete(self, key: str):
    """Deletes document with an id that equals to `key`."""

get(key) abstractmethod

Gets document with an id that equals to key.

Source code in solnlib/modular_input/checkpointer.py 
53
54
55
@abstractmethod
def get(self, key: str) -> dict:
    """Gets document with an id that equals to `key`."""

update(key, state) abstractmethod

Updates document with an id that equals to key and state as document data.

Source code in solnlib/modular_input/checkpointer.py 
44
45
46
47
@abstractmethod
def update(self, key: str, state: Any):
    """Updates document with an id that equals to `key` and `state` as
    document data."""

CheckpointerException

Bases: Exception

Source code in solnlib/modular_input/checkpointer.py 
37
38
class CheckpointerException(Exception):
    pass

FileCheckpointer

Bases: Checkpointer

File checkpointer.

Use file to save modular input checkpoint.

Examples:

>>> from solnlib.modular_input import checkpointer
>>> ck = checkpointer.FileCheckpointer('/opt/splunk/var/...')
>>> ck.update(...)
>>> ck.get(...)
Source code in solnlib/modular_input/checkpointer.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
class FileCheckpointer(Checkpointer):
    """File checkpointer.

    Use file to save modular input checkpoint.

    Examples:
        >>> from solnlib.modular_input import checkpointer
        >>> ck = checkpointer.FileCheckpointer('/opt/splunk/var/...')
        >>> ck.update(...)
        >>> ck.get(...)
    """

    def __init__(self, checkpoint_dir: str):
        """Initializes FileCheckpointer.

        Arguments:
            checkpoint_dir: Checkpoint directory.
        """
        warnings.warn(
            "FileCheckpointer is deprecated, please use KVStoreCheckpointer",
            stacklevel=2,
        )
        self._checkpoint_dir = checkpoint_dir

    def encode_key(self, key):
        return base64.b64encode(key.encode()).decode()

    def update(self, key, state):
        file_name = op.join(self._checkpoint_dir, self.encode_key(key))
        with open(file_name + "_new", "w") as fp:
            json.dump(state, fp)

        if op.exists(file_name):
            try:
                os.remove(file_name)
            except OSError:
                pass

        os.rename(file_name + "_new", file_name)

    def batch_update(self, states):
        for state in states:
            self.update(state["_key"], state["state"])

    def get(self, key):
        file_name = op.join(self._checkpoint_dir, self.encode_key(key))
        try:
            with open(file_name) as fp:
                return json.load(fp)
        except (OSError, ValueError):
            return None

    def delete(self, key):
        file_name = op.join(self._checkpoint_dir, self.encode_key(key))
        try:
            os.remove(file_name)
        except OSError:
            pass

__init__(checkpoint_dir)

Initializes FileCheckpointer.

Parameters:

Name Type Description Default
checkpoint_dir str

Checkpoint directory.

 required
Source code in solnlib/modular_input/checkpointer.py 
218
219
220
221
222
223
224
225
226
227
228
def __init__(self, checkpoint_dir: str):
    """Initializes FileCheckpointer.

    Arguments:
        checkpoint_dir: Checkpoint directory.
    """
    warnings.warn(
        "FileCheckpointer is deprecated, please use KVStoreCheckpointer",
        stacklevel=2,
    )
    self._checkpoint_dir = checkpoint_dir

batch_update(states)

Source code in solnlib/modular_input/checkpointer.py 
246
247
248
def batch_update(self, states):
    for state in states:
        self.update(state["_key"], state["state"])

delete(key)

Source code in solnlib/modular_input/checkpointer.py 
258
259
260
261
262
263
def delete(self, key):
    file_name = op.join(self._checkpoint_dir, self.encode_key(key))
    try:
        os.remove(file_name)
    except OSError:
        pass

encode_key(key)

Source code in solnlib/modular_input/checkpointer.py 
230
231
def encode_key(self, key):
    return base64.b64encode(key.encode()).decode()

get(key)

Source code in solnlib/modular_input/checkpointer.py 
250
251
252
253
254
255
256
def get(self, key):
    file_name = op.join(self._checkpoint_dir, self.encode_key(key))
    try:
        with open(file_name) as fp:
            return json.load(fp)
    except (OSError, ValueError):
        return None

update(key, state)

Source code in solnlib/modular_input/checkpointer.py 
233
234
235
236
237
238
239
240
241
242
243
244
def update(self, key, state):
    file_name = op.join(self._checkpoint_dir, self.encode_key(key))
    with open(file_name + "_new", "w") as fp:
        json.dump(state, fp)

    if op.exists(file_name):
        try:
            os.remove(file_name)
        except OSError:
            pass

    os.rename(file_name + "_new", file_name)

KVStoreCheckpointer

Bases: Checkpointer

KVStore checkpointer.

Use KVStore to save modular input checkpoint.

More information about KV Store in Splunk is here.

Examples:

>>> from solnlib.modular_input import checkpointer
>>> checkpoint = checkpointer.KVStoreCheckpointer(
        "unique_addon_checkpoints",
        "session_key",
        "unique_addon"
    )
>>> checkpoint.update("input_1", {"timestamp": 1638043093})
>>> checkpoint.get("input_1")
>>> # returns {"timestamp": 1638043093}
Source code in solnlib/modular_input/checkpointer.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
class KVStoreCheckpointer(Checkpointer):
    """KVStore checkpointer.

    Use KVStore to save modular input checkpoint.

    More information about KV Store in Splunk is
    [here](https://dev.splunk.com/enterprise/docs/developapps/manageknowledge/kvstore/aboutkvstorecollections).

    Examples:
        >>> from solnlib.modular_input import checkpointer
        >>> checkpoint = checkpointer.KVStoreCheckpointer(
                "unique_addon_checkpoints",
                "session_key",
                "unique_addon"
            )
        >>> checkpoint.update("input_1", {"timestamp": 1638043093})
        >>> checkpoint.get("input_1")
        >>> # returns {"timestamp": 1638043093}
    """

    def __init__(
        self,
        collection_name: str,
        session_key: str,
        app: str,
        owner: Optional[str] = "nobody",
        scheme: Optional[str] = None,
        host: Optional[str] = None,
        port: Optional[int] = None,
        **context: Any,
    ):
        """Initializes KVStoreCheckpointer.

        Arguments:
            collection_name: Collection name of kvstore checkpointer.
            session_key: Splunk access token.
            app: App name of namespace.
            owner: (optional) Owner of namespace, default is `nobody`.
            scheme: (optional) The access scheme, default is None.
            host: (optional) The host name, default is None.
            port: (optional) The port number, default is None.
            context: Other configurations for Splunk rest client.

        Raises:
            binding.HTTPError: HTTP error different from 404, for example 503
                when KV Store is initializing and not ready to serve requests.
            CheckpointerException: If init KV Store checkpointer failed.
        """
        try:
            if not context.get("pool_connections"):
                context["pool_connections"] = 5
            if not context.get("pool_maxsize"):
                context["pool_maxsize"] = 5
            self._collection_data = _utils.get_collection_data(
                collection_name,
                session_key,
                app,
                owner,
                scheme,
                host,
                port,
                {"state": "string"},
                **context,
            )
        except KeyError:
            raise CheckpointerException("Get KV Store checkpointer failed.")

    @utils.retry(exceptions=[binding.HTTPError])
    def update(self, key: str, state: Any) -> None:
        """Updates document with an id that equals to `key` and `state` as
        document data.

        Arguments:
            key: `id` of the document to update.
            state: Document data to update. It can be integer, string,
                or a dict, or anything that can be an argument to `json.dumps`.

        Raises:
            binding.HTTPError: when an error occurred in Splunk, for example,
                when Splunk is restarting and KV Store is not yet initialized.
        """
        record = {"_key": key, "state": json.dumps(state)}
        self._collection_data.batch_save(record)

    @utils.retry(exceptions=[binding.HTTPError])
    def batch_update(self, states: Iterable[Dict[str, Any]]) -> None:
        """Updates multiple documents.

        Arguments:
            states: Iterable that contains documents to update. Document should
                be a dict with at least "state" key.

        Raises:
            binding.HTTPError: when an error occurred in Splunk, for example,
                when Splunk is restarting and KV Store is not yet initialized.
        """
        for state in states:
            state["state"] = json.dumps(state["state"])
        self._collection_data.batch_save(*states)

    @utils.retry(exceptions=[binding.HTTPError])
    def get(self, key: str) -> Optional[Any]:
        """Gets document with an id that equals to `key`.

        Arguments:
            key: `id` of the document to get.

        Raises:
            binding.HTTPError: When an error occurred in Splunk (not 404 code),
                can be 503 code, when Splunk is restarting and KV Store is not
                yet initialized.

        Returns:
            Document data under `key` or `None` in case of no data.
        """
        try:
            record = self._collection_data.query_by_id(key)
        except binding.HTTPError as e:
            if e.status != 404:
                logging.error(f"Get checkpoint failed: {traceback.format_exc()}.")
                raise
            return None
        return json.loads(record["state"])

    @utils.retry(exceptions=[binding.HTTPError])
    def delete(self, key: str) -> None:
        """Deletes document with an id that equals to `key`.

        Arguments:
            key: `id` of the document to delete.

        Raises:
            binding.HTTPError: When an error occurred in Splunk (not 404 code),
                can be 503 code, when Splunk is restarting and KV Store is not
                yet initialized.
        """
        try:
            self._collection_data.delete_by_id(key)
        except binding.HTTPError as e:
            if e.status != 404:
                logging.error(f"Delete checkpoint failed: {traceback.format_exc()}.")
                raise

__init__(collection_name, session_key, app, owner='nobody', scheme=None, host=None, port=None, **context)

Initializes KVStoreCheckpointer.

Parameters:

Name Type Description Default
collection_name str

Collection name of kvstore checkpointer.

 required
session_key str

Splunk access token.

 required
app str

App name of namespace.

 required
owner Optional[str]

(optional) Owner of namespace, default is nobody.

 'nobody'
scheme Optional[str]

(optional) The access scheme, default is None.

 None
host Optional[str]

(optional) The host name, default is None.

 None
port Optional[int]

(optional) The port number, default is None.

 None
context Any

Other configurations for Splunk rest client.

 {}

Raises:

Type Description
binding.HTTPError

HTTP error different from 404, for example 503 when KV Store is initializing and not ready to serve requests.
CheckpointerException

If init KV Store checkpointer failed.
Source code in solnlib/modular_input/checkpointer.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def __init__(
    self,
    collection_name: str,
    session_key: str,
    app: str,
    owner: Optional[str] = "nobody",
    scheme: Optional[str] = None,
    host: Optional[str] = None,
    port: Optional[int] = None,
    **context: Any,
):
    """Initializes KVStoreCheckpointer.

    Arguments:
        collection_name: Collection name of kvstore checkpointer.
        session_key: Splunk access token.
        app: App name of namespace.
        owner: (optional) Owner of namespace, default is `nobody`.
        scheme: (optional) The access scheme, default is None.
        host: (optional) The host name, default is None.
        port: (optional) The port number, default is None.
        context: Other configurations for Splunk rest client.

    Raises:
        binding.HTTPError: HTTP error different from 404, for example 503
            when KV Store is initializing and not ready to serve requests.
        CheckpointerException: If init KV Store checkpointer failed.
    """
    try:
        if not context.get("pool_connections"):
            context["pool_connections"] = 5
        if not context.get("pool_maxsize"):
            context["pool_maxsize"] = 5
        self._collection_data = _utils.get_collection_data(
            collection_name,
            session_key,
            app,
            owner,
            scheme,
            host,
            port,
            {"state": "string"},
            **context,
        )
    except KeyError:
        raise CheckpointerException("Get KV Store checkpointer failed.")

batch_update(states)

Updates multiple documents.

Parameters:

Name Type Description Default
states Iterable[Dict[str, Any]]

Iterable that contains documents to update. Document should be a dict with at least “state” key.

 required

Raises:

Type Description
binding.HTTPError

when an error occurred in Splunk, for example, when Splunk is restarting and KV Store is not yet initialized.
Source code in solnlib/modular_input/checkpointer.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@utils.retry(exceptions=[binding.HTTPError])
def batch_update(self, states: Iterable[Dict[str, Any]]) -> None:
    """Updates multiple documents.

    Arguments:
        states: Iterable that contains documents to update. Document should
            be a dict with at least "state" key.

    Raises:
        binding.HTTPError: when an error occurred in Splunk, for example,
            when Splunk is restarting and KV Store is not yet initialized.
    """
    for state in states:
        state["state"] = json.dumps(state["state"])
    self._collection_data.batch_save(*states)

delete(key)

Deletes document with an id that equals to key.

Parameters:

Name Type Description Default
key str

id of the document to delete.

 required

Raises:

Type Description
binding.HTTPError

When an error occurred in Splunk (not 404 code), can be 503 code, when Splunk is restarting and KV Store is not yet initialized.
Source code in solnlib/modular_input/checkpointer.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
@utils.retry(exceptions=[binding.HTTPError])
def delete(self, key: str) -> None:
    """Deletes document with an id that equals to `key`.

    Arguments:
        key: `id` of the document to delete.

    Raises:
        binding.HTTPError: When an error occurred in Splunk (not 404 code),
            can be 503 code, when Splunk is restarting and KV Store is not
            yet initialized.
    """
    try:
        self._collection_data.delete_by_id(key)
    except binding.HTTPError as e:
        if e.status != 404:
            logging.error(f"Delete checkpoint failed: {traceback.format_exc()}.")
            raise

get(key)

Gets document with an id that equals to key.

Parameters:

Name Type Description Default
key str

id of the document to get.

 required

Raises:

Type Description
binding.HTTPError

When an error occurred in Splunk (not 404 code), can be 503 code, when Splunk is restarting and KV Store is not yet initialized.

Returns:

Type Description
Optional[Any]

Document data under key or None in case of no data.
Source code in solnlib/modular_input/checkpointer.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
@utils.retry(exceptions=[binding.HTTPError])
def get(self, key: str) -> Optional[Any]:
    """Gets document with an id that equals to `key`.

    Arguments:
        key: `id` of the document to get.

    Raises:
        binding.HTTPError: When an error occurred in Splunk (not 404 code),
            can be 503 code, when Splunk is restarting and KV Store is not
            yet initialized.

    Returns:
        Document data under `key` or `None` in case of no data.
    """
    try:
        record = self._collection_data.query_by_id(key)
    except binding.HTTPError as e:
        if e.status != 404:
            logging.error(f"Get checkpoint failed: {traceback.format_exc()}.")
            raise
        return None
    return json.loads(record["state"])

update(key, state)

Updates document with an id that equals to key and state as document data.

Parameters:

Name Type Description Default
key str

id of the document to update.

 required
state Any

Document data to update. It can be integer, string, or a dict, or anything that can be an argument to json.dumps.

 required

Raises:

Type Description
binding.HTTPError

when an error occurred in Splunk, for example, when Splunk is restarting and KV Store is not yet initialized.
Source code in solnlib/modular_input/checkpointer.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@utils.retry(exceptions=[binding.HTTPError])
def update(self, key: str, state: Any) -> None:
    """Updates document with an id that equals to `key` and `state` as
    document data.

    Arguments:
        key: `id` of the document to update.
        state: Document data to update. It can be integer, string,
            or a dict, or anything that can be an argument to `json.dumps`.

    Raises:
        binding.HTTPError: when an error occurred in Splunk, for example,
            when Splunk is restarting and KV Store is not yet initialized.
    """
    record = {"_key": key, "state": json.dumps(state)}
    self._collection_data.batch_save(record)