Texte vergleichen

Finde den Unterschied zwischen zwei Textdateien

Live-Editor

Gleiches ausblenden

Zeilenumbruch aus

Ansicht

Vergleichsgenauigkeit

Syntaxhervorhebung

Diffchecker Desktop Der sicherste Weg, Diffchecker zu nutzen. Hol dir die Desktop-App: Deine Diffs verlassen nie deinen Computer!Desktop holen

Untitled diff

Erstellt vor 10 JahrenDiff läuft nie ab

49 Entfernungen

Zeilen
Gesamt
Entfernt

Zeichen
Gesamt
Entfernt

Um diese Funktion weiterhin zu nutzen, aktualisiere auf Diffchecker Pro Preise anzeigen

204 Zeilen

25 Hinzufügungen

Zeilen
Gesamt
Hinzugefügt

Zeichen
Gesamt
Hinzugefügt

Um diese Funktion weiterhin zu nutzen, aktualisiere auf Diffchecker Pro Preise anzeigen

194 Zeilen

#!/usr/bin/env python

import httplib

try:

import simplejson as json

except ImportError:

import json

import os

import sys

from urlparse import urljoin

try:

import requests

except ImportError:

raise ImportError('Missing dependency "requests". Do ``pip install requests``.')

raise ImportError('Missing dependency requests. Do ``pip install requests``.')

try:

import yaml

except ImportError:

raise ImportError('Missing dependency "pyyaml". Do ``pip install pyyaml``.')

raise ImportError('Missing dependency pyyaml. Do ``pip install pyyaml``.')

# ST2 configuration

ST2_CONFIG_FILE = './config.yaml'

ST2_API_BASE_URL = 'http://localhost:9101/v1'

ST2_AUTH_BASE_URL = 'http://localhost:9100'

ST2_USERNAME = None

ST2_PASSWORD = None

ST2_AUTH_TOKEN = None

ST2_AUTH_PATH = 'tokens'

ST2_WEBHOOKS_PATH = 'webhooks/st2/'

ST2_TRIGGERS_PATH = 'triggertypes/'

ST2_TRIGGERTYPE_PACK = 'e_nagios'

ST2_TRIGGERTYPE_PACK = 'nagios'

ST2_TRIGGERTYPE_NAME = 'service_state_change'

ST2_TRIGGERTYPE_NAME = 'service-state-change'

ST2_TRIGGERTYPE_REF = '.'.join([ST2_TRIGGERTYPE_PACK, ST2_TRIGGERTYPE_NAME])

STATE_MESSAGE = {

'OK': 'All is well on the Western front.',

'WARNING': 'We gots a warning yo!',

'UNKNOWN': 'It be unknown...',

'CRITICAL': 'Critical!'

}

REGISTERED_WITH_ST2 = False

OK_CODES = [httplib.OK, httplib.CREATED, httplib.ACCEPTED, httplib.CONFLICT]

STATE_MESSAGE = {

'OK': '[OK] Service/Host is back to normal.',

'WARNING': '[WARNING] Service/Host warning alert!',

'UNKNOWN': '[UNKNOWN] Service/Host is in the unknown state.',

'CRITICAL': '[CRITICAL] Pay attention immediately!'

}

def _create_trigger_type():

try:

url = _get_st2_triggers_url()

payload = {

'name': ST2_TRIGGERTYPE_NAME,

'pack': ST2_TRIGGERTYPE_PACK,

'description': 'Trigger type for nagios event handler.'

}

# sys.stdout.write('POST: %s: Body: %s\n' % (url, payload))

headers = {}

headers['Content-Type'] = 'application/json; charset=utf-8'

if ST2_AUTH_TOKEN:

headers['X-Auth-Token'] = ST2_AUTH_TOKEN

post_resp = requests.post(url, data=json.dumps(payload), headers=headers, verify=False)

post_resp = requests.post(url, data=json.dumps(payload), headers=headers)

except:

sys.stderr.write('Unable to register trigger type with st2.')

raise

else:

status = post_resp.status_code

if status not in OK_CODES:

sys.stderr.write('Failed to register trigger type with st2. HTTP_CODE: %d\n' %

status)

raise

else:

sys.stdout.write('Registered trigger type with st2.\n')

def _get_auth_url():

return urljoin(ST2_AUTH_BASE_URL, ST2_AUTH_PATH)

def _get_auth_token():

global ST2_AUTH_TOKEN

auth_url = _get_auth_url()

try:

resp = requests.post(auth_url, json.dumps({'ttl': 5 * 60}),

auth=(ST2_USERNAME, ST2_PASSWORD), verify=False)

auth=(ST2_USERNAME, ST2_PASSWORD))

except:

sys.stderr.write('Cannot get auth token from st2. Trying unauthed.')

raise Exception('Cannot get auth token from st2. Will try unauthed.')

else:

if resp.status_code not in OK_CODES:

raise Exception("Cannot authenticate: %s\n" % resp.text)

ST2_AUTH_TOKEN = resp.json()['token']

def _register_with_st2():

global REGISTERED_WITH_ST2

try:

url = urljoin(_get_st2_triggers_url(), ST2_TRIGGERTYPE_REF)

# sys.stdout.write('GET: %s\n' % url)

if not ST2_AUTH_TOKEN:

_get_auth_token()

if ST2_AUTH_TOKEN:

get_resp = requests.get(url, headers={'X-Auth-Token': ST2_AUTH_TOKEN}, verify=False)

get_resp = requests.get(url, headers={'X-Auth-Token': ST2_AUTH_TOKEN})

else:

get_resp = requests.get(url, verify=False)

get_resp = requests.get(url)

if get_resp.status_code != httplib.OK:

_create_trigger_type()

else:

body = json.loads(get_resp.text)

if len(body) == 0:

_create_trigger_type()

except:

raise

else:

REGISTERED_WITH_ST2 = True

def _get_st2_triggers_url():

url = urljoin(ST2_API_BASE_URL, ST2_TRIGGERS_PATH)

return url

def _get_st2_webhooks_url():

url = urljoin(ST2_API_BASE_URL, ST2_WEBHOOKS_PATH)

return url

def _post_event_to_st2(url, body):

headers = {}

headers['X-ST2-Integration'] = 'nagios.'

headers['Content-Type'] = 'application/json; charset=utf-8'

if ST2_AUTH_TOKEN:

headers['X-Auth-Token'] = ST2_AUTH_TOKEN

try:

sys.stdout.write('POST: url: %s, body: %s\n' % (url, body))

# sys.stdout.write('POST: url: %s, body: %s\n' % (url, body))

r = requests.post(url, data=json.dumps(body), headers=headers, verify=False)

r = requests.post(url, data=json.dumps(body), headers=headers)

except:

sys.stderr.write('Cannot connect to st2 endpoint.')

else:

status = r.status_code

if status not in OK_CODES:

sys.stderr.write('Failed posting nagios event to st2. HTTP_CODE: %d\n' % status)

else:

sys.stdout.write('Sent nagios event to st2. HTTP_CODE: %d\n' % status)

def _get_payload(host, service, event_id, state, state_id, state_type, attempt):

def _get_payload(host, service, event_id, state, state_type, attempt):

payload = {}

payload['host'] = host

payload['service'] = service

payload['event_id'] = event_id

payload['state'] = state

payload['state_id'] = state_id

payload['state_type'] = state_type

payload['attempt'] = attempt

payload['msg'] = STATE_MESSAGE.get(state, 'Undefined state.')

return payload

def main(args):

event_id = args[1]

service = args[2]

state = args[3]

state_id = args[4]

state_type = args[4]

state_type = args[5]

attempt = args[5]

attempt = args[6]

host = args[6]

host = args[7]

payload = _get_payload(host, service, event_id, state, state_type, attempt)

body = {}

body['trigger'] = ST2_TRIGGERTYPE_REF

body['payload'] = _get_payload(host, service, event_id, state, state_id, state_type, attempt)

body['payload'] = payload

_post_event_to_st2(_get_st2_webhooks_url(), body)

if __name__ == '__main__':

if len(sys.argv) > 1:

st2_config_file = sys.argv[1]

sys.argv.pop(1)

else:

sys.stderr.write('Error: config file missing.\n')

sys.stderr.write('Usage: %s ST2_CONFIG_FILE\n' % sys.argv[0])

sys.exit(2)

try:

if not os.path.exists(st2_config_file):

if not os.path.exists(ST2_CONFIG_FILE):

sys.stderr.write('Configuration file not found. Exiting.\n')

sys.exit(1)

with open(st2_config_file) as f:

with open(ST2_CONFIG_FILE) as f:

config = yaml.safe_load(f)

ST2_USERNAME = config['st2_username']

ST2_PASSWORD = config['st2_password']

ST2_API_BASE_URL = config['st2_api_base_url']

ST2_AUTH_BASE_URL = config['st2_auth_base_url']

if not REGISTERED_WITH_ST2:

_register_with_st2()

except Exception as e:

except:

sys.stderr.write(

sys.stderr.write('Failed registering with st2. Won\'t post event.\n')

'Failed registering with st2. Won\'t post event.\n%s' % e)

# import traceback

# traceback.print_exc()

else:

main(sys.argv)

Gespeicherte Diffs

Originaltext

Datei öffnen

#!/usr/bin/env python

import httplib
try:
    import simplejson as json
except ImportError:
    import json
import os
import sys
from urlparse import urljoin

try:
    import requests
except ImportError:
    raise ImportError('Missing dependency "requests". Do ``pip install requests``.')

try:
    import yaml
except ImportError:
    raise ImportError('Missing dependency "pyyaml". Do ``pip install pyyaml``.')

# ST2 configuration

ST2_API_BASE_URL = 'http://localhost:9101/v1'
ST2_AUTH_BASE_URL = 'http://localhost:9100'
ST2_USERNAME = None
ST2_PASSWORD = None
ST2_AUTH_TOKEN = None

ST2_AUTH_PATH = 'tokens'
ST2_WEBHOOKS_PATH = 'webhooks/st2/'
ST2_TRIGGERS_PATH = 'triggertypes/'
ST2_TRIGGERTYPE_PACK = 'e_nagios'
ST2_TRIGGERTYPE_NAME = 'service_state_change'
ST2_TRIGGERTYPE_REF = '.'.join([ST2_TRIGGERTYPE_PACK, ST2_TRIGGERTYPE_NAME])

REGISTERED_WITH_ST2 = False

OK_CODES = [httplib.OK, httplib.CREATED, httplib.ACCEPTED, httplib.CONFLICT]

STATE_MESSAGE = {
    'OK': '[OK] Service/Host is back to normal.',
    'WARNING': '[WARNING] Service/Host warning alert!',
    'UNKNOWN': '[UNKNOWN] Service/Host is in the unknown state.',
    'CRITICAL': '[CRITICAL] Pay attention immediately!'
}

def _create_trigger_type():
    try:
        url = _get_st2_triggers_url()
        payload = {
            'name': ST2_TRIGGERTYPE_NAME,
            'pack': ST2_TRIGGERTYPE_PACK,
            'description': 'Trigger type for nagios event handler.'
        }
        # sys.stdout.write('POST: %s: Body: %s\n' % (url, payload))
        headers = {}
        headers['Content-Type'] = 'application/json; charset=utf-8'

if ST2_AUTH_TOKEN:
            headers['X-Auth-Token'] = ST2_AUTH_TOKEN

post_resp = requests.post(url, data=json.dumps(payload), headers=headers, verify=False)
    except:
        sys.stderr.write('Unable to register trigger type with st2.')
        raise
    else:
        status = post_resp.status_code
        if status not in OK_CODES:
            sys.stderr.write('Failed to register trigger type with st2. HTTP_CODE: %d\n' %
                             status)
            raise
        else:
            sys.stdout.write('Registered trigger type with st2.\n')

def _get_auth_url():
    return urljoin(ST2_AUTH_BASE_URL, ST2_AUTH_PATH)

def _get_auth_token():
    global ST2_AUTH_TOKEN
    auth_url = _get_auth_url()
    try:
        resp = requests.post(auth_url, json.dumps({'ttl': 5 * 60}),
                             auth=(ST2_USERNAME, ST2_PASSWORD), verify=False)
    except:
        sys.stderr.write('Cannot get auth token from st2. Trying unauthed.')
    else:
        if resp.status_code not in OK_CODES:
            raise Exception("Cannot authenticate: %s\n" % resp.text)

ST2_AUTH_TOKEN = resp.json()['token']

def _register_with_st2():
    global REGISTERED_WITH_ST2
    try:
        url = urljoin(_get_st2_triggers_url(), ST2_TRIGGERTYPE_REF)
        # sys.stdout.write('GET: %s\n' % url)
        if not ST2_AUTH_TOKEN:
            _get_auth_token()

if ST2_AUTH_TOKEN:
            get_resp = requests.get(url, headers={'X-Auth-Token': ST2_AUTH_TOKEN}, verify=False)
        else:
            get_resp = requests.get(url, verify=False)

if get_resp.status_code != httplib.OK:
            _create_trigger_type()
        else:
            body = json.loads(get_resp.text)
            if len(body) == 0:
                _create_trigger_type()
    except:
        raise
    else:
        REGISTERED_WITH_ST2 = True

def _get_st2_triggers_url():
    url = urljoin(ST2_API_BASE_URL, ST2_TRIGGERS_PATH)
    return url

def _get_st2_webhooks_url():
    url = urljoin(ST2_API_BASE_URL, ST2_WEBHOOKS_PATH)
    return url

def _post_event_to_st2(url, body):
    headers = {}
    headers['X-ST2-Integration'] = 'nagios.'
    headers['Content-Type'] = 'application/json; charset=utf-8'
    if ST2_AUTH_TOKEN:
        headers['X-Auth-Token'] = ST2_AUTH_TOKEN
    try:
        sys.stdout.write('POST: url: %s, body: %s\n' % (url, body))
        r = requests.post(url, data=json.dumps(body), headers=headers, verify=False)
    except:
        sys.stderr.write('Cannot connect to st2 endpoint.')
    else:
        status = r.status_code
        if status not in OK_CODES:
            sys.stderr.write('Failed posting nagios event to st2. HTTP_CODE: %d\n' % status)
        else:
            sys.stdout.write('Sent nagios event to st2. HTTP_CODE: %d\n' % status)

def _get_payload(host, service, event_id, state, state_id, state_type, attempt):
    payload = {}
    payload['host'] = host
    payload['service'] = service
    payload['event_id'] = event_id
    payload['state'] = state
    payload['state_id'] = state_id
    payload['state_type'] = state_type
    payload['attempt'] = attempt
    payload['msg'] = STATE_MESSAGE.get(state, 'Undefined state.')
    return payload

def main(args):
    event_id = args[1]
    service = args[2]
    state = args[3]
    state_id = args[4]
    state_type = args[5]
    attempt = args[6]
    host = args[7]
    body = {}
    body['trigger'] = ST2_TRIGGERTYPE_REF
    body['payload'] = _get_payload(host, service, event_id, state, state_id, state_type, attempt)
    _post_event_to_st2(_get_st2_webhooks_url(), body)

if __name__ == '__main__':
    if len(sys.argv) > 1:
        st2_config_file = sys.argv[1]
        sys.argv.pop(1)
    else:
        sys.stderr.write('Error: config file missing.\n')
        sys.stderr.write('Usage: %s ST2_CONFIG_FILE\n' % sys.argv[0])
        sys.exit(2)

try:
        if not os.path.exists(st2_config_file):
            sys.stderr.write('Configuration file not found. Exiting.\n')
            sys.exit(1)

with open(st2_config_file) as f:
            config = yaml.safe_load(f)
            ST2_USERNAME = config['st2_username']
            ST2_PASSWORD = config['st2_password']
            ST2_API_BASE_URL = config['st2_api_base_url']
            ST2_AUTH_BASE_URL = config['st2_auth_base_url']

if not REGISTERED_WITH_ST2:
            _register_with_st2()
    except Exception as e:
        sys.stderr.write(
            'Failed registering with st2. Won\'t post event.\n%s' % e)
#        import traceback
#        traceback.print_exc()
    else:
        main(sys.argv)

Bearbeitung

Datei öffnen

#!/usr/bin/env python

import httplib
try:
    import simplejson as json
except ImportError:
    import json
import os
import sys
from urlparse import urljoin

try:
    import requests
except ImportError:
    raise ImportError('Missing dependency requests. Do ``pip install requests``.')

try:
    import yaml
except ImportError:
    raise ImportError('Missing dependency pyyaml. Do ``pip install pyyaml``.')

# ST2 configuration
ST2_CONFIG_FILE = './config.yaml'

ST2_API_BASE_URL = 'http://localhost:9101/v1'
ST2_AUTH_BASE_URL = 'http://localhost:9100'
ST2_USERNAME = None
ST2_PASSWORD = None
ST2_AUTH_TOKEN = None

ST2_AUTH_PATH = 'tokens'
ST2_WEBHOOKS_PATH = 'webhooks/st2/'
ST2_TRIGGERS_PATH = 'triggertypes/'
ST2_TRIGGERTYPE_PACK = 'nagios'
ST2_TRIGGERTYPE_NAME = 'service-state-change'
ST2_TRIGGERTYPE_REF = '.'.join([ST2_TRIGGERTYPE_PACK, ST2_TRIGGERTYPE_NAME])

STATE_MESSAGE = {
    'OK': 'All is well on the Western front.',
    'WARNING': 'We gots a warning yo!',
    'UNKNOWN': 'It be unknown...',
    'CRITICAL': 'Critical!'
}

REGISTERED_WITH_ST2 = False

OK_CODES = [httplib.OK, httplib.CREATED, httplib.ACCEPTED, httplib.CONFLICT]

if ST2_AUTH_TOKEN:
            headers['X-Auth-Token'] = ST2_AUTH_TOKEN

post_resp = requests.post(url, data=json.dumps(payload), headers=headers)
    except:
        sys.stderr.write('Unable to register trigger type with st2.')
        raise
    else:
        status = post_resp.status_code
        if status not in OK_CODES:
            sys.stderr.write('Failed to register trigger type with st2. HTTP_CODE: %d\n' %
                             status)
            raise
        else:
            sys.stdout.write('Registered trigger type with st2.\n')

def _get_auth_url():
    return urljoin(ST2_AUTH_BASE_URL, ST2_AUTH_PATH)

def _get_auth_token():
    global ST2_AUTH_TOKEN
    auth_url = _get_auth_url()
    try:
        resp = requests.post(auth_url, json.dumps({'ttl': 5 * 60}),
                             auth=(ST2_USERNAME, ST2_PASSWORD))
    except:
        raise Exception('Cannot get auth token from st2. Will try unauthed.')
    else:
        ST2_AUTH_TOKEN = resp.json()['token']

if ST2_AUTH_TOKEN:
            get_resp = requests.get(url, headers={'X-Auth-Token': ST2_AUTH_TOKEN})
        else:
            get_resp = requests.get(url)

def _get_st2_triggers_url():
    url = urljoin(ST2_API_BASE_URL, ST2_TRIGGERS_PATH)
    return url

def _get_st2_webhooks_url():
    url = urljoin(ST2_API_BASE_URL, ST2_WEBHOOKS_PATH)
    return url

def _post_event_to_st2(url, body):
    headers = {}
    headers['X-ST2-Integration'] = 'nagios.'
    headers['Content-Type'] = 'application/json; charset=utf-8'
    if ST2_AUTH_TOKEN:
        headers['X-Auth-Token'] = ST2_AUTH_TOKEN
    try:
        # sys.stdout.write('POST: url: %s, body: %s\n' % (url, body))
        r = requests.post(url, data=json.dumps(body), headers=headers)
    except:
        sys.stderr.write('Cannot connect to st2 endpoint.')
    else:
        status = r.status_code
        if status not in OK_CODES:
            sys.stderr.write('Failed posting nagios event to st2. HTTP_CODE: %d\n' % status)
        else:
            sys.stdout.write('Sent nagios event to st2. HTTP_CODE: %d\n' % status)

def _get_payload(host, service, event_id, state, state_type, attempt):
    payload = {}
    payload['host'] = host
    payload['service'] = service
    payload['event_id'] = event_id
    payload['state'] = state
    payload['state_type'] = state_type
    payload['attempt'] = attempt
    payload['msg'] = STATE_MESSAGE.get(state, 'Undefined state.')
    return payload

def main(args):
    event_id = args[1]
    service = args[2]
    state = args[3]
    state_type = args[4]
    attempt = args[5]
    host = args[6]

payload = _get_payload(host, service, event_id, state, state_type, attempt)
    body = {}
    body['trigger'] = ST2_TRIGGERTYPE_REF
    body['payload'] = payload
    _post_event_to_st2(_get_st2_webhooks_url(), body)

if __name__ == '__main__':
    try:
        if not os.path.exists(ST2_CONFIG_FILE):
            sys.stderr.write('Configuration file not found. Exiting.\n')
            sys.exit(1)

with open(ST2_CONFIG_FILE) as f:
            config = yaml.safe_load(f)
            ST2_USERNAME = config['st2_username']
            ST2_PASSWORD = config['st2_password']
            ST2_API_BASE_URL = config['st2_api_base_url']
            ST2_AUTH_BASE_URL = config['st2_auth_base_url']

if not REGISTERED_WITH_ST2:
            _register_with_st2()
    except:
        sys.stderr.write('Failed registering with st2. Won\'t post event.\n')
    else:
        main(sys.argv)