Hi, you can try by API.
here is script, with make api call and save OS version to file.
Then you can send result as a metric and make present it on dashboard:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import argparse
import csv
import sys
from collections import Counter
from typing import Dict, List, Optional, Any, Tuple
import requests
from requests.adapters import HTTPAdapter, Retry
# === CONFIGURATION (edit as needed) ===
DT_BASE_URL = "https://dynatrace.example.com/e/xxx" # Your base URL (Managed/SaaS) with /e/<env-id>
DT_API_TOKEN = "API TOKEN" # Token with entities.read permission
# Default output files:
DEFAULT_SUMMARY_CSV = "hosts_by_os_summary.csv"
DEFAULT_HOSTS_CSV = "hosts_by_os_hosts.csv"
# === Platform normalization based on osType/osVersion ===
def normalize_platform(os_type: Optional[str], os_version: Optional[str]) -> str:
ot = (os_type or "").strip().upper()
ov = (os_version or "").strip().lower()
if ot == "WINDOWS":
return "Windows"
if ot == "AIX":
return "AIX"
if ot == "SOLARIS":
return "Solaris"
if ot in ("ZOS", "Z/OS"):
return "z/OS"
if ot == "LINUX":
if "oracle" in ov or " ol " in f" {ov} ":
return "Oracle Linux"
if "red hat" in ov or "rhel" in ov:
return "Red Hat Enterprise Linux"
if "ubuntu" in ov:
return "Ubuntu"
if "debian" in ov:
return "Debian"
if "suse" in ov or "sles" in ov:
return "SUSE Linux"
if "centos" in ov:
return "CentOS"
if "almalinux" in ov:
return "AlmaLinux"
if "rocky" in ov:
return "Rocky Linux"
return "Linux (Other)"
if ot:
return ot.title()
return "Unknown"
def make_session(api_token: str, verify: Optional[bool or str]) -> requests.Session:
"""Create a requests session with retries and headers."""
s = requests.Session()
retries = Retry(
total=5,
backoff_factor=0.5,
status_forcelist=(429, 500, 502, 503, 504),
allowed_methods=("GET", "HEAD"),
)
s.mount("https://", HTTPAdapter(max_retries=retries))
s.mount("http://", HTTPAdapter(max_retries=retries))
s.headers.update({
"Authorization": f"Api-Token {api_token}",
"Accept": "application/json; charset=utf-8",
})
# Keep TLS verification setting on the session
s.verify = verify
return s
def fetch_hosts(session: requests.Session,
base_url: str,
page_size: int = 500,
management_zone: Optional[str] = None) -> List[Dict[str, Any]]:
"""
Fetch all HOST entities with osType, osVersion (and detectedName via properties)
"""
url = f"{base_url.rstrip('/')}/api/v2/entities"
entity_selector = 'type("HOST")'
if management_zone:
entity_selector += f',managementZone("{management_zone}")'
params = {
"entitySelector": entity_selector,
# NOTE: displayName must NOT be included in fields=; properties.* are allowed.
"fields": "properties.osType,properties.osVersion,properties.detectedName",
"pageSize": page_size,
}
results: List[Dict[str, Any]] = []
next_key: Optional[str] = None
while True:
if next_key:
resp = session.get(url, params={"nextPageKey": next_key})
else:
resp = session.get(url, params=params)
resp.raise_for_status()
data = resp.json()
entities = data.get("entities", [])
results.extend(entities)
next_key = data.get("nextPageKey")
if not next_key:
break
return results
def summarize_and_flatten(entities: List[Dict[str, Any]]) -> Tuple[Counter, List[Dict[str, str]]]:
"""
Returns:
- Counter per platform
- rows for detailed CSV (entityId, name, osType, osVersion, platform)
"""
summary = Counter()
hosts_rows: List[Dict[str, str]] = []
for e in entities:
props = e.get("properties", {}) or {}
os_type = props.get("osType")
os_version = props.get("osVersion")
platform = normalize_platform(os_type, os_version)
# Prefer displayName, then detectedName, then entityId
display_name = e.get("displayName") or props.get("detectedName") or e.get("entityId")
summary[platform] += 1
hosts_rows.append({
"entityId": e.get("entityId", ""),
"name": display_name or "",
"osType": os_type or "",
"osVersion": os_version or "",
"platform": platform,
})
# Sort rows by platform then name
hosts_rows.sort(key=lambda r: (r["platform"].lower(), r["name"].lower()))
return summary, hosts_rows
def write_summary_csv(path: str, summary: Counter) -> None:
"""Write platform summary to CSV."""
with open(path, "w", newline="", encoding="utf-8") as f:
w = csv.writer(f, delimiter=";")
w.writerow(["platform", "count"])
for platform, cnt in summary.most_common():
w.writerow([platform, cnt])
def write_hosts_csv(path: str, rows: List[Dict[str, str]]) -> None:
"""Write detailed host list to CSV."""
fieldnames = ["entityId", "name", "osType", "osVersion", "platform"]
with open(path, "w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=fieldnames, delimiter=";")
w.writeheader()
for r in rows:
w.writerow(r)
def print_console_table(summary: Counter) -> None:
"""Pretty-print a small summary table to console."""
total = sum(summary.values())
if total == 0:
print("No hosts returned (0).")
return
width = max((len(k) for k in summary.keys()), default=8)
print("\nHosts by operating system (summary):")
print(f"{'Platform'.ljust(width)} | {'Count':>6} | {'Share':>6}")
print("-" * (width + 22))
for platform, cnt in summary.most_common():
share = f"{(cnt/total*100):.1f}%"
print(f"{platform.ljust(width)} | {str(cnt).rjust(6)} | {share.rjust(6)}")
print("-" * (width + 22))
print(f"{'TOTAL'.ljust(width)} | {str(total).rjust(6)} | 100.0%\n")
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Count hosts by OS (Dynatrace API v2).")
p.add_argument("--base-url", default=DT_BASE_URL,
help="Base environment URL, e.g., https://dynatrace.example.com/e/<env-id>")
p.add_argument("--token", default=DT_API_TOKEN,
help="API token with entities.read permission")
p.add_argument("--mz", default=None,
help="(optional) Management Zone name to filter")
p.add_argument("--page-size", type=int, default=500,
help="Page size (default 500)")
p.add_argument("--out-summary", default=DEFAULT_SUMMARY_CSV,
help=f"CSV file for platform summary (default {DEFAULT_SUMMARY_CSV})")
p.add_argument("--out-hosts", default=DEFAULT_HOSTS_CSV,
help=f"CSV file for host list (default {DEFAULT_HOSTS_CSV})")
p.add_argument("--verify", default=None,
help="Path to trusted CA file (e.g., /etc/ssl/certs/rootCA.pem). "
"If omitted, system CA will be used.")
p.add_argument("--insecure", action="store_true",
help="Disable TLS verification (NOT recommended).")
return p.parse_args()
def main() -> int:
args = parse_args()
# Basic param validation
if not args.base_url or "example.com" in args.base_url:
print("Please set --base-url or adjust DT_BASE_URL in the script.", file=sys.stderr)
return 2
if not args.token or args.token.endswith("REPLACE_ME"):
print("Please set --token or adjust DT_API_TOKEN in the script.", file=sys.stderr)
return 2
# TLS verification
verify: Optional[bool or str]
if args.insecure:
verify = False
elif args.verify:
verify = args.verify
else:
verify = True # system CA
# Create session
session = make_session(args.token, verify)
# Fetch hosts
try:
entities = fetch_hosts(
session=session,
base_url=args.base_url,
page_size=args.page_size,
management_zone=args.mz,
)
except requests.HTTPError as e:
print(f"HTTP error: {e} | body: {getattr(e.response, 'text', '')}", file=sys.stderr)
return 1
except requests.RequestException as e:
print(f"✖ Network error: {e}", file=sys.stderr)
return 1
# Summarize
summary, host_rows = summarize_and_flatten(entities)
# Print console table
print_console_table(summary)
# Write CSV outputs
try:
write_summary_csv(args.out_summary, summary)
write_hosts_csv(args.out_hosts, host_rows)
except OSError as e:
print(f"File write error: {e}", file=sys.stderr)
return 1
print(f"Summary CSV saved: {args.out_summary}")
print(f"Hosts CSV saved: {args.out_hosts}")
return 0
if __name__ == "__main__":
sys.exit(main())
