Reporting on Collectors with specific preferred collector set

Summary: Will O'Reilly is seeking guidance on identifying resources linked to a specific collector set in their inventory report. Their current setup outputs particular collectors but does not indicate if they are auto-balanced. This inquiry arises as they plan to introduce new collectors and decommission old ones, ensuring that no resources are statically set to older collectors to avoid monitoring gaps upon shutdown.
I'm looking for a way to find out what resources I have that have a specific collector set rather than auto-balanced.
I have an inventory report setup to output collector details. However, this is spitting out specific collectors rather than telling me it's auto-balanced.
The context behind this is that I'm adding new collectors with a view to decommissioning old collectors. I need to check that nothing has the old collectors set statically so that when I shut them down, there's no gap in monitoring.
Will O'Reilly
Posted 5 months ago · Edited 5 months ago·Last reply 5 months ago
1 comment
Jan Garaj
·5 months ago
I did it a few weeks ago via API. Author is LLM, so be careful and don't blame me, when something will be wrong:
#!/usr/bin/env python3
"""
LogicMonitor Preferred Collector Unsetter

This script uses the LogicMonitor v3 SDK to:
1. Discover all resources where preferredCollectorId > 0
2. Patch these resources to set preferredCollectorId to 0 (auto-assign)

Two-phase workflow:
  View phase  - discover matching resources, print statistics, wait for approval
  Apply phase - update resources concurrently using a configurable worker pool

Requirements:
- logicmonitor-sdk library: pip install logicmonitor-sdk
- LogicMonitor API credentials (Access ID, Access Key, Company name)

Examples:
  # View phase only (default – no changes made)
  python unset-preffered-collector.py

  # Apply changes after view phase confirmation (5 workers, no limit)
  python unset-preffered-collector.py --apply

  # Test run: view + apply only 10 resources
  python unset-preffered-collector.py --apply --limit 10

  # Increase worker count
  python unset-preffered-collector.py --apply --workers 10
"""

import os
import sys
import time
import argparse
import concurrent.futures
from typing import List, Dict, Any, Optional, Tuple, Callable
from dataclasses import dataclass

from dotenv import load_dotenv
import logicmonitor_sdk
from logicmonitor_sdk.rest import ApiException


# ---------------------------------------------------------------------------
# Data containers
# ---------------------------------------------------------------------------

@dataclass
class ResourceUpdate:
    device_id: int
    display_name: str
    preferred_collector_id: int


@dataclass
class UpdateResult:
    resource: ResourceUpdate
    success: bool
    error: Optional[str] = None


# ---------------------------------------------------------------------------
# LogicMonitor client
# ---------------------------------------------------------------------------

class LogicMonitorClient:
    def __init__(self, company: str, access_id: str, access_key: str):
        configuration = logicmonitor_sdk.Configuration()
        configuration.company = company
        configuration.access_id = access_id
        configuration.access_key = access_key

        self.api_client = logicmonitor_sdk.ApiClient(configuration)
        self.lm_api = logicmonitor_sdk.LMApi(self.api_client)

    # ------------------------------------------------------------------
    # Rate-limit-aware retry helper
    # ------------------------------------------------------------------

    @staticmethod
    def _call_with_retry(fn: Callable, *args, max_retries: int = 5, **kwargs) -> Any:
        """
        Call fn(*args, **kwargs) and retry on HTTP 429 (Too Many Requests).

        On a 429 response the LM API returns three headers:
          x-rate-limit-limit     – total allowed requests per window
          x-rate-limit-remaining – requests left in current window
          x-rate-limit-window    – window duration in seconds

        We sleep for the window duration (plus a small buffer) before retrying.
        """
        for attempt in range(1, max_retries + 1):
            try:
                return fn(*args, **kwargs)
            except ApiException as e:
                if e.status != 429:
                    raise

                headers = e.headers or {}
                try:
                    window = int(headers.get('x-rate-limit-window', 60))
                    limit = int(headers.get('x-rate-limit-limit', '?') or 60)
                    remaining = int(headers.get('x-rate-limit-remaining', 0))
                except (ValueError, TypeError):
                    window, limit, remaining = 60, '?', 0

                wait = window + 2
                print(
                    f"  [rate-limit] 429 received (limit={limit}, remaining={remaining}, "
                    f"window={window}s) – waiting {wait}s before retry {attempt}/{max_retries}..."
                )
                time.sleep(wait)

        raise RuntimeError(f"Still getting 429 after {max_retries} retries")

    # ------------------------------------------------------------------
    # Device fetching
    # ------------------------------------------------------------------

    def _fetch_page(self, offset: int, size: int) -> List[Any]:
        try:
            response = self._call_with_retry(
                self.lm_api.get_device_list,
                filter="preferredCollectorId>0",
                size=size,
                offset=offset,
            )
            return response.items or []
        except ApiException as e:
            print(f"  API error at offset={offset}: {e}")
            return []

    def get_devices_with_preferred_collector(self, workers: int = 5) -> List[Any]:
        """
        Fetch all devices where preferredCollectorId > 0.
        Pages are fetched in parallel using a thread pool for speed.
        """
        size = 1000
        print("Fetching devices with preferredCollectorId > 0...")

        first_page = self._fetch_page(0, size)
        if not first_page:
            return []

        try:
            probe = self._call_with_retry(
                self.lm_api.get_device_list,
                filter="preferredCollectorId>0",
                size=1,
                offset=0,
            )
            total = probe.total or 0
        except ApiException:
            total = 0

        if total <= size:
            print(f"  Fetched {len(first_page)} devices (single page)")
            return first_page

        remaining_offsets = list(range(size, total, size))
        print(f"  Total devices reported: {total} – fetching {len(remaining_offsets)} more page(s) with {workers} workers...")

        all_pages: List[List[Any]] = [None] * (len(remaining_offsets) + 1)  # type: ignore[list-item]
        all_pages[0] = first_page

        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
            future_map = {
                executor.submit(self._fetch_page, off, size): idx + 1
                for idx, off in enumerate(remaining_offsets)
            }
            done = 0
            for future in concurrent.futures.as_completed(future_map):
                idx = future_map[future]
                all_pages[idx] = future.result()
                done += 1
                if done % 10 == 0 or done == len(remaining_offsets):
                    fetched = sum(len(p) for p in all_pages if p is not None)
                    print(f"  Pages done: {done}/{len(remaining_offsets)}  ({fetched} devices fetched)")

        devices: List[Any] = []
        for page in all_pages:
            if page:
                devices.extend(page)

        print(f"  Fetch complete: {len(devices)} devices total")
        return devices

    # ------------------------------------------------------------------
    # Patch helper
    # ------------------------------------------------------------------

    def unset_preferred_collector(self, device_id: int) -> bool:
        body = {'preferredCollectorId': 0}
        try:
            self._call_with_retry(
                self.lm_api.api_client.call_api,
                f'/device/devices/{device_id}',
                'PATCH',
                query_params=[('opType', 'replace')],
                body=body,
                response_type='Device',
                auth_settings=['LMv1'],
            )
            return True
        except Exception as e:
            raise RuntimeError(str(e)) from e

    # ------------------------------------------------------------------
    # View phase
    # ------------------------------------------------------------------

    def discover_resources_to_update(
        self,
        limit: Optional[int] = None,
        workers: int = 5,
    ) -> Tuple[List[ResourceUpdate], Dict[str, Any]]:
        """
        Phase 1: discover all resources where preferredCollectorId > 0.
        Returns the list of pending updates and a statistics dict.
        """
        print()
        all_devices = self.get_devices_with_preferred_collector(workers=workers)
        print(f"\nTotal devices with preferred collector: {len(all_devices)}")

        updates: List[ResourceUpdate] = [
            ResourceUpdate(
                device_id=device.id,
                display_name=device.display_name,
                preferred_collector_id=device.preferred_collector_id,
            )
            for device in all_devices
        ]

        stats: Dict[str, Any] = {
            "total_devices": len(all_devices),
            "to_update": len(updates),
        }

        if limit is not None:
            updates = updates[:limit]

        return updates, stats

    # ------------------------------------------------------------------
    # Apply phase
    # ------------------------------------------------------------------

    def apply_update(self, update: ResourceUpdate) -> UpdateResult:
        try:
            self.unset_preferred_collector(update.device_id)
            return UpdateResult(resource=update, success=True)
        except Exception as e:
            return UpdateResult(resource=update, success=False, error=str(e))

    def apply_updates(
        self,
        updates: List[ResourceUpdate],
        workers: int = 5,
    ) -> List[UpdateResult]:
        """
        Phase 2: apply updates concurrently using a thread pool.
        """
        results: List[UpdateResult] = []
        total = len(updates)

        print(f"\nApplying {total} update(s) using {workers} worker(s)...\n")

        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
            future_to_update = {
                executor.submit(self.apply_update, u): u for u in updates
            }

            done_count = 0
            for future in concurrent.futures.as_completed(future_to_update):
                result = future.result()
                results.append(result)
                done_count += 1

                if result.success:
                    print(
                        f"  [{done_count}/{total}] OK  {result.resource.display_name}"
                        f"  (collector {result.resource.preferred_collector_id} → 0)"
                    )
                else:
                    print(
                        f"  [{done_count}/{total}] ERR {result.resource.display_name}"
                        f"  (collector {result.resource.preferred_collector_id}) – {result.error}"
                    )

        return results


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def print_view_report(updates: List[ResourceUpdate], stats: Dict[str, Any], limit: Optional[int]) -> None:
    print("\n" + "=" * 80)
    print("VIEW PHASE REPORT")
    print("=" * 80)
    print(f"  Total devices with preferred collector : {stats.get('total_devices', 0)}")
    print(f"  Resources eligible to update           : {stats.get('to_update', 0)}")

    if limit is not None and stats.get('to_update', 0) > limit:
        print(f"  Applying limit                         : {limit} (of {stats['to_update']} eligible)")
        print(f"  Will update                            : {len(updates)}")
    else:
        print(f"  Will update                            : {len(updates)}")

    if updates:
        print(f"\n  Resources queued for update ({len(updates)}):")
        print(f"  {'Resource':<60} {'Preferred Collector ID'}")
        print("  " + "-" * 85)
        for u in updates:
            print(f"  {u.display_name:<60} {u.preferred_collector_id}")

    print("=" * 80)


def print_apply_summary(results: List[UpdateResult]) -> None:
    succeeded = [r for r in results if r.success]
    failed = [r for r in results if not r.success]

    print("\n" + "=" * 80)
    print("APPLY SUMMARY")
    print("=" * 80)
    print(f"  Succeeded : {len(succeeded)}")
    print(f"  Failed    : {len(failed)}")

    if failed:
        print("\n  Failed resources:")
        for r in failed:
            print(f"    - {r.resource.display_name} (ID: {r.resource.device_id}) – {r.error}")

    print("=" * 80)


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

def main() -> None:
    load_dotenv()

    parser = argparse.ArgumentParser(
        description='Set preferredCollectorId to 0 on all resources where it is > 0',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )

    parser.add_argument('--company',
                        help='LogicMonitor company name (subdomain). Also: LM_COMPANY env var')
    parser.add_argument('--access-id',
                        help='API Access ID. Also: LM_ACCESS_ID env var')
    parser.add_argument('--access-key',
                        help='API Access Key. Also: LM_ACCESS_KEY env var')
    parser.add_argument('--apply',
                        action='store_true',
                        help='After the view phase, prompt for confirmation and apply changes')
    parser.add_argument('--limit',
                        type=int,
                        default=None,
                        metavar='N',
                        help='Only update up to N resources (useful for testing)')
    parser.add_argument('--workers',
                        type=int,
                        default=5,
                        metavar='N',
                        help='Number of parallel workers for updates (default: 5)')
    parser.add_argument('--yes',
                        action='store_true',
                        help='Skip confirmation prompt and apply immediately (use with --apply)')

    args = parser.parse_args()

    company = args.company or os.getenv('LM_COMPANY')
    access_id = args.access_id or os.getenv('LM_ACCESS_ID')
    access_key = args.access_key or os.getenv('LM_ACCESS_KEY')

    if not company:
        print("Error: company required via --company or LM_COMPANY env var")
        sys.exit(1)
    if not access_id:
        print("Error: access ID required via --access-id or LM_ACCESS_ID env var")
        sys.exit(1)
    if not access_key:
        print("Error: access key required via --access-key or LM_ACCESS_KEY env var")
        sys.exit(1)

    try:
        print(f"Connecting to LogicMonitor: {company}.logicmonitor.com")
        client = LogicMonitorClient(
            company=company,
            access_id=access_id,
            access_key=access_key,
        )

        # ---- View phase ------------------------------------------------
        print("\n" + "=" * 80)
        print("VIEW PHASE – discovering resources to update")
        print("=" * 80)

        updates, stats = client.discover_resources_to_update(
            limit=args.limit,
            workers=args.workers,
        )

        print_view_report(updates, stats, args.limit)

        if not updates:
            print("\nNo resources to update. Exiting.")
            sys.exit(0)

        if not args.apply:
            print("\nView phase complete. Re-run with --apply to apply changes.")
            sys.exit(0)

        # ---- Confirmation ----------------------------------------------
        if not args.yes:
            print(f"\nAbout to update {len(updates)} resource(s) using {args.workers} worker(s).")
            answer = input("Proceed? [y/N] ").strip().lower()
            if answer not in ('y', 'yes'):
                print("Aborted.")
                sys.exit(0)

        # ---- Apply phase -----------------------------------------------
        results = client.apply_updates(updates, workers=args.workers)
        print_apply_summary(results)

        failed_count = sum(1 for r in results if not r.success)
        sys.exit(1 if failed_count > 0 else 0)

    except KeyboardInterrupt:
        print("\nInterrupted.")
        sys.exit(130)
    except Exception as e:
        print(f"Error: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)


if __name__ == "__main__":
    main()
1	SUYASH GAIKWAD
2	FF Fujitsu Fazeer
3	EN Ernest Nganza
4	AM Abhishek M
5	NA Nazmeen Ali
6	DK Deepak Kumar