Come join our live training webinar every other Wednesday at 11am PST and hear LogicMonitor experts explain best practices and answer common questions. We understand these are uncertain times, and we are here to help!
You can use LogicMonitor Push Metrics REST API to programmatically ingest metrics for multiple instances associated with a single resource and DataSource.
The Push Metrics REST API requires LMv1 API Token Authentication.
The metric ingestion endpoint is used to push metrics to one or more instances associated with a DataSource and resource. Only one resource and DataSource is supported per request. If this is the first time metrics are being pushed to a resource, this request will result in a new DataSource and optionally a new resource (depending on parameters).
Use the following mandatory and optional fields in your request.
The following Python 3 example monitors CPU utilization.
#!/usr/bin/env python3 import time, os, sys import hmac, hashlib, base64 import requests import json import psutil #-------------------------------------- # Constants - Do not change #-------------------------------------- resource_path = '/metric/ingest' headers = { 'Content-Type': 'application/json' } #--------------------------------------- # Change Values below as per your setup #--------------------------------------- def get_params(): # Account info # Your account name Company = "ABC" # User account Access Id for which it must have permissions to manage resources and manage LogicModules AccessId = "6wDEvh73taGMPP43SSmF" AccessKey = 'Q)cfY2w8u[G97jkz5^P8{HT)g-uq[6+E8LL%Mb{^' url = "https://"+Company+".logicmonitor.com/rest"+ resource_path return url, AccessId, AccessKey #-------------------------------------------- # Function to send metric data to LM Platform #-------------------------------------------- def send_metrics(timestamp, body): url, AccessId, AccessKey = get_params() req_var = "POST" + str(timestamp) + body +resource_path; signature = base64.b64encode(bytes(hmac.new( bytes(AccessKey, 'latin-1'), bytes(req_var, 'latin-1'), digestmod=hashlib.sha256 ).hexdigest(), 'latin-1')).decode('latin-1') auth = "LMv1 "+AccessId+ ":"+ signature+":"+str(timestamp) headers['Authorization'] = auth try: response = requests.post(url, verify=True, headers=headers, data=body, params={"create":'true'}) if response.status_code != 202: print('Failed to send metric. Error:', response.status_code, response.text) else: print("SUCCESS :",response.text) except Exception as e: print("Unable to connect. Error: ", e) #-------------------------------------------- # Prepare REST payload #-------------------------------------------- def prepare_request_body(metric, timestamp, data_value): return json.dumps({ "resourceName": metric["device_name"], "resourceIds": { "system.displayname": metric["device_name"], "system.ips": metric["device_ip"] }, "dataSource": metric["data_source"], "dataSourceDisplayName": metric["data_source"], "instances": [ { "instanceName": metric["instance"], "instanceDisplayName": metric["instance"], "instanceProperties": { "version": "1", }, "dataPoints": [ { "dataPointName": metric["data_point"], "dataPointType": "GAUGE", "dataPointAggregationType": "sum", "values": { str(timestamp//1000): data_value } }, ] }, ] }).replace("'", '"') #------------------ #==== MAIN ======= #------------------ if __name__ == "__main__": my_metric = {} my_metric["device_name"] = os.uname()[1] my_metric["device_ip"] = "192.168.1.1" my_metric["data_source"] = "CPU" my_metric["instance"] = "cpu-1" my_metric["data_point"] = "cpu_utilization" while True: timestamp = int(time.time()*1000) data_value = psutil.cpu_percent() body = prepare_request_body(my_metric, timestamp, data_value) send_metrics(timestamp, body) time.sleep(10)
In This Article