from dynatrace import Dynatrace from dynatrace import TOO_MANY_REQUESTS_WAIT from dynatrace.environment_v2.tokens_api import SCOPE_METRICS_READ, SCOPE_METRICS_INGEST from dynatrace.configuration_v1.credential_vault import PublicCertificateCredentials from dynatrace.environment_v2.settings import SettingsObject, SettingsObjectCreate from datetime import datetime, timedelta # Create a Dynatrace client dt = Dynatrace("environment_url", "api_token") # Create a client that handles too many requests (429) # dt = Dynatrace("environment_url", "api_token", too_many_requests_strategy=TOO_MANY_REQUESTS_WAIT ) # Create a client that automatically retries on errors, up to 5 times, with a 1 second delay between retries # dt = Dynatrace("environment_url", "api_token", retries=5, retry_delay_ms=1000 ) # Create a client with a custom HTTP timeout of 10 seconds # dt = Dynatrace("environment_url", "api_token", timeout=10 ) # Get all hosts and some properties for entity in dt.entities.list('type("HOST")', fields="properties.memoryTotal,properties.monitoringMode"): print(entity.entity_id, entity.display_name, entity.properties) # Get idle CPU for all hosts for metric in dt.metrics.query("builtin:host.cpu.idle", resolution="Inf"): print(metric) # Print dimensions, timestamp and values for the AWS Billing Metric for metric in dt.metrics.query("ext:cloud.aws.billing.estimatedChargesByRegionCurrency"): for data in metric.data: for timestamp, value in zip(data.timestamps, data.values): print(data.dimensions, timestamp, value) # Get all ActiveGates for ag in dt.activegates.list(): print(ag) # Get metric descriptions for all host metrics for m in dt.metrics.list("builtin:host.*"): print(m) # Delete endpoints that contain the word test for plugin in dt.plugins.list(): # This could also be dt.get_endpoints(plugin.id) for endpoint in plugin.endpoints: if "test" in endpoint.name: endpoint.delete(plugin.id) # Prints dashboard ID, owner and number of tiles for dashboard in dt.dashboards.list(): full_dashboard = dashboard.get_full_dashboard() print(full_dashboard.id, dashboard.owner, len(full_dashboard.tiles)) # Delete API Tokens that haven't been used for more than 3 months for token in dt.tokens.list(fields="+lastUsedDate,+scopes"): if token.last_used_date and token.last_used_date < datetime.now() - timedelta(days=90): print(f"Deleting token! {token}, last used date: {token.last_used_date}") # Create an API Token that can read and ingest metrics new_token = dt.tokens.create("metrics_token", scopes=[SCOPE_METRICS_READ, SCOPE_METRICS_INGEST]) print(new_token.token) # Upload a public PEM certificate to the Credential Vault with open("ca.pem", "r") as f: ca_cert = f.read() my_cred = PublicCertificateCredentials( name="my_cred", description="my_cred description", scope="EXTENSION", owner_access_only=False, certificate=ca_cert, password="", credential_type="PUBLIC_CERTIFICATE", certificate_format="PEM" ) r = dt.credentials.post(my_cred) print(r.id) # Create a new settings 2.0 object settings_value = { "enabled": True, "summary": "DT API TEST 1", "queryDefinition": { "type": "METRIC_KEY", "metricKey": "netapp.ontap.node.fru.state", "aggregation": "AVG", "entityFilter": { "dimensionKey": "dt.entity.netapp_ontap:fru", "conditions": [], }, "dimensionFilter": [], }, "modelProperties": { "type": "STATIC_THRESHOLD", "threshold": 100.0, "alertOnNoData": False, "alertCondition": "BELOW", "violatingSamples": 3, "samples": 5, "dealertingSamples": 5, }, "eventTemplate": { "title": "OnTap {dims:type} {dims:fru_id} is in Error State", "description": "OnTap field replaceable unit (FRU) {dims:type} with id {dims:fru_id} on node {dims:node} in cluster {dims:cluster} is in an error state.\n", "eventType": "RESOURCE", "davisMerge": True, "metadata": [], }, "eventEntityDimensionKey": "dt.entity.netapp_ontap:fru", } settings_object = SettingsObjectCreate(schema_id="builtin:anomaly-detection.metric-events", value=settings_value, scope="environment") dt.settings.create_object(validate_only=False, body=settings_object)