Skip to content

hec_config.py

__all__ = ['HECConfig'] module-attribute

HECConfig

HTTP Event Collector configuration.

Source code in solnlib/hec_config.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 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
class HECConfig:
    """HTTP Event Collector configuration."""

    input_type = "http"

    def __init__(
        self,
        session_key: str,
        scheme: str = None,
        host: str = None,
        port: int = None,
        **context: dict
    ):
        """Initializes HECConfig.

        Arguments:
            session_key: Splunk access token.
            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.
        """
        self._rest_client = rest_client.SplunkRestClient(
            session_key,
            "splunk_httpinput",
            scheme=scheme,
            host=host,
            port=port,
            **context
        )

    @retry(exceptions=[binding.HTTPError])
    def get_settings(self) -> dict:
        """Get http data input global settings.

        Returns:
            HTTP global settings, for example:

                {
                    'enableSSL': 1,
                    'disabled': 0,
                    'useDeploymentServer': 0,
                    'port': 8088
                }
        """

        return self._do_get_input(self.input_type).content

    @retry(exceptions=[binding.HTTPError])
    def update_settings(self, settings: dict):
        """Update http data input global settings.

        Arguments:
            settings: HTTP global settings.
        """

        res = self._do_get_input(self.input_type)
        res.update(**settings)

    @retry(exceptions=[binding.HTTPError])
    def create_input(self, name: str, stanza: dict) -> dict:
        """Create http data input.

        Arguments:
            name: HTTP data input name.
            stanza: Data input stanza content.

        Returns:
            Created input.

        Examples:
           >>> from solnlib.hec_config import HECConfig
           >>> hec = HECConfig(session_key)
           >>> hec.create_input('my_hec_data_input',
                                {'index': 'main', 'sourcetype': 'hec'})
        """

        res = self._rest_client.inputs.create(name, self.input_type, **stanza)
        return res.content

    @retry(exceptions=[binding.HTTPError])
    def update_input(self, name: str, stanza: dict):
        """Update http data input.

        It will create if the data input doesn't exist.

        Arguments:
            name: HTTP data input name.
            stanza: Data input stanza.

        Examples:
           >>> from solnlib import HEConfig
           >>> hec = HECConfig(session_key)
           >>> hec.update_input('my_hec_data_input',
                                {'index': 'main', 'sourcetype': 'hec2'})
        """

        res = self._do_get_input(name)
        if res is None:
            return self.create_input(name, stanza)
        res.update(**stanza)

    @retry(exceptions=[binding.HTTPError])
    def delete_input(self, name: str):
        """Delete http data input.

        Arguments:
            name: HTTP data input name.
        """

        try:
            self._rest_client.inputs.delete(name, self.input_type)
        except KeyError:
            pass

    @retry(exceptions=[binding.HTTPError])
    def get_input(self, name: str) -> dict:
        """Get http data input.

        Arguments:
            name: HTTP event collector data input name.

        Returns:
            HTTP event collector data input config dict.
        """

        res = self._do_get_input(name)
        if res:
            return res.content
        else:
            return None

    def _do_get_input(self, name):
        try:
            return self._rest_client.inputs[(name, self.input_type)]
        except KeyError:
            return None

    @retry(exceptions=[binding.HTTPError])
    def get_limits(self) -> dict:
        """Get HTTP input limits.

        Returns:
            HTTP input limits.
        """

        return self._rest_client.confs["limits"]["http_input"].content

    @retry(exceptions=[binding.HTTPError])
    def set_limits(self, limits: dict):
        """Set HTTP input limits.

        Arguments:
            limits: HTTP input limits.
        """

        res = self._rest_client.confs["limits"]["http_input"]
        res.submit(limits)

input_type = 'http' class-attribute instance-attribute

__init__(session_key, scheme=None, host=None, port=None, **context)

Initializes HECConfig.

Parameters:

Name Type Description Default
session_key str

Splunk access token.

 required
scheme str

(optional) The access scheme, default is None.

 None
host str

(optional) The host name, default is None.

 None
port int

(optional) The port number, default is None.

 None
context dict

Other configurations for Splunk rest client.

 {}
Source code in solnlib/hec_config.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(
    self,
    session_key: str,
    scheme: str = None,
    host: str = None,
    port: int = None,
    **context: dict
):
    """Initializes HECConfig.

    Arguments:
        session_key: Splunk access token.
        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.
    """
    self._rest_client = rest_client.SplunkRestClient(
        session_key,
        "splunk_httpinput",
        scheme=scheme,
        host=host,
        port=port,
        **context
    )

create_input(name, stanza)

Create http data input.

Parameters:

Name Type Description Default
name str

HTTP data input name.

 required
stanza dict

Data input stanza content.

 required

Returns:

Type Description
dict

Created input.

Examples:

>>> from solnlib.hec_config import HECConfig
>>> hec = HECConfig(session_key)
>>> hec.create_input('my_hec_data_input',
                     {'index': 'main', 'sourcetype': 'hec'})
Source code in solnlib/hec_config.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@retry(exceptions=[binding.HTTPError])
def create_input(self, name: str, stanza: dict) -> dict:
    """Create http data input.

    Arguments:
        name: HTTP data input name.
        stanza: Data input stanza content.

    Returns:
        Created input.

    Examples:
       >>> from solnlib.hec_config import HECConfig
       >>> hec = HECConfig(session_key)
       >>> hec.create_input('my_hec_data_input',
                            {'index': 'main', 'sourcetype': 'hec'})
    """

    res = self._rest_client.inputs.create(name, self.input_type, **stanza)
    return res.content

delete_input(name)

Delete http data input.

Parameters:

Name Type Description Default
name str

HTTP data input name.

 required
Source code in solnlib/hec_config.py 
127
128
129
130
131
132
133
134
135
136
137
138
@retry(exceptions=[binding.HTTPError])
def delete_input(self, name: str):
    """Delete http data input.

    Arguments:
        name: HTTP data input name.
    """

    try:
        self._rest_client.inputs.delete(name, self.input_type)
    except KeyError:
        pass

get_input(name)

Get http data input.

Parameters:

Name Type Description Default
name str

HTTP event collector data input name.

 required

Returns:

Type Description
dict

HTTP event collector data input config dict.
Source code in solnlib/hec_config.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
@retry(exceptions=[binding.HTTPError])
def get_input(self, name: str) -> dict:
    """Get http data input.

    Arguments:
        name: HTTP event collector data input name.

    Returns:
        HTTP event collector data input config dict.
    """

    res = self._do_get_input(name)
    if res:
        return res.content
    else:
        return None

get_limits()

Get HTTP input limits.

Returns:

Type Description
dict

HTTP input limits.
Source code in solnlib/hec_config.py 
163
164
165
166
167
168
169
170
171
@retry(exceptions=[binding.HTTPError])
def get_limits(self) -> dict:
    """Get HTTP input limits.

    Returns:
        HTTP input limits.
    """

    return self._rest_client.confs["limits"]["http_input"].content

get_settings()

Get http data input global settings.

Returns:

Type Description
dict

HTTP global settings, for example:

{ ‘enableSSL’: 1, ‘disabled’: 0, ‘useDeploymentServer’: 0, ‘port’: 8088 }
Source code in solnlib/hec_config.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@retry(exceptions=[binding.HTTPError])
def get_settings(self) -> dict:
    """Get http data input global settings.

    Returns:
        HTTP global settings, for example:

            {
                'enableSSL': 1,
                'disabled': 0,
                'useDeploymentServer': 0,
                'port': 8088
            }
    """

    return self._do_get_input(self.input_type).content

set_limits(limits)

Set HTTP input limits.

Parameters:

Name Type Description Default
limits dict

HTTP input limits.

 required
Source code in solnlib/hec_config.py 
173
174
175
176
177
178
179
180
181
182
@retry(exceptions=[binding.HTTPError])
def set_limits(self, limits: dict):
    """Set HTTP input limits.

    Arguments:
        limits: HTTP input limits.
    """

    res = self._rest_client.confs["limits"]["http_input"]
    res.submit(limits)

update_input(name, stanza)

Update http data input.

It will create if the data input doesn’t exist.

Parameters:

Name Type Description Default
name str

HTTP data input name.

 required
stanza dict

Data input stanza.

 required

Examples:

>>> from solnlib import HEConfig
>>> hec = HECConfig(session_key)
>>> hec.update_input('my_hec_data_input',
                     {'index': 'main', 'sourcetype': 'hec2'})
Source code in solnlib/hec_config.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@retry(exceptions=[binding.HTTPError])
def update_input(self, name: str, stanza: dict):
    """Update http data input.

    It will create if the data input doesn't exist.

    Arguments:
        name: HTTP data input name.
        stanza: Data input stanza.

    Examples:
       >>> from solnlib import HEConfig
       >>> hec = HECConfig(session_key)
       >>> hec.update_input('my_hec_data_input',
                            {'index': 'main', 'sourcetype': 'hec2'})
    """

    res = self._do_get_input(name)
    if res is None:
        return self.create_input(name, stanza)
    res.update(**stanza)

update_settings(settings)

Update http data input global settings.

Parameters:

Name Type Description Default
settings dict

HTTP global settings.

 required
Source code in solnlib/hec_config.py 
73
74
75
76
77
78
79
80
81
82
@retry(exceptions=[binding.HTTPError])
def update_settings(self, settings: dict):
    """Update http data input global settings.

    Arguments:
        settings: HTTP global settings.
    """

    res = self._do_get_input(self.input_type)
    res.update(**settings)