Skip to content

Plugin System

BackendPlugin()

Bases: PluginBase, BackendInterface, MetricSink

Base class for backend plugins

Source code in tracelet/core/plugins.py 
 99
100
101
def __init__(self):
    self._config = {}
    self._active = False

can_handle_type(metric_type)

Check if this backend can handle the metric type

Source code in tracelet/core/plugins.py 
107
108
109
110
def can_handle_type(self, metric_type: MetricType) -> bool:
    """Check if this backend can handle the metric type"""
    # By default, backends handle all metric types
    return True

get_sink_id()

Return sink ID for orchestrator integration

Source code in tracelet/core/plugins.py 
103
104
105
def get_sink_id(self) -> str:
    """Return sink ID for orchestrator integration"""
    return f"backend_{self.get_metadata().name}"

CollectorPlugin()

Bases: PluginBase, CollectorInterface

Base class for collector plugins

Source code in tracelet/core/plugins.py 
116
117
118
119
def __init__(self):
    self._config = {}
    self._active = False
    self._collection_interval = 60  # seconds

PluginBase

Bases: ABC

Base class for all plugins

get_metadata() abstractmethod classmethod

Return plugin metadata

Source code in tracelet/core/plugins.py 
69
70
71
72
73
@classmethod
@abstractmethod
def get_metadata(cls) -> PluginMetadata:
    """Return plugin metadata"""
    pass

get_status() abstractmethod

Get plugin status

Source code in tracelet/core/plugins.py 
90
91
92
93
@abstractmethod
def get_status(self) -> dict[str, Any]:
    """Get plugin status"""
    pass

initialize(config) abstractmethod

Initialize the plugin with configuration

Source code in tracelet/core/plugins.py 
75
76
77
78
@abstractmethod
def initialize(self, config: dict[str, Any]):
    """Initialize the plugin with configuration"""
    pass

start() abstractmethod

Start the plugin

Source code in tracelet/core/plugins.py 
80
81
82
83
@abstractmethod
def start(self):
    """Start the plugin"""
    pass

stop() abstractmethod

Stop the plugin

Source code in tracelet/core/plugins.py 
85
86
87
88
@abstractmethod
def stop(self):
    """Stop the plugin"""
    pass

PluginInfo(metadata, module_path, class_name, state=PluginState.DISCOVERED, instance=None, error=None) dataclass

Information about a discovered plugin

PluginManager(plugin_paths=None, use_default_paths=True)

Manages plugin discovery, loading, and lifecycle

Source code in tracelet/core/plugins.py 
125
126
127
128
129
130
131
132
133
def __init__(self, plugin_paths: Optional[list[str]] = None, use_default_paths: bool = True):
    self.plugin_paths = plugin_paths or []
    self.plugins: dict[str, PluginInfo] = {}
    self._plugin_instances: dict[str, PluginBase] = {}
    self._dependency_graph: dict[str, set[str]] = {}

    # Add default plugin paths unless explicitly disabled
    if use_default_paths:
        self._add_default_paths()

discover_plugins()

Discover all available plugins

Source code in tracelet/core/plugins.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def discover_plugins(self) -> list[PluginInfo]:
    """Discover all available plugins"""
    discovered = []

    for path in self.plugin_paths:
        if os.path.isdir(path):
            discovered.extend(self._discover_in_directory(path))
        elif path.endswith(".py"):
            plugin = self._discover_in_file(path)
            if plugin:
                discovered.append(plugin)

    # Update internal registry
    for plugin in discovered:
        self.plugins[plugin.metadata.name] = plugin
        logger.info(f"Discovered plugin: {plugin.metadata.name} v{plugin.metadata.version}")

    return discovered

get_plugin_instance(plugin_name)

Get an initialized plugin instance

Source code in tracelet/core/plugins.py 
361
362
363
def get_plugin_instance(self, plugin_name: str) -> Optional[PluginBase]:
    """Get an initialized plugin instance"""
    return self._plugin_instances.get(plugin_name)

get_plugins_by_type(plugin_type)

Get all plugins of a specific type

Source code in tracelet/core/plugins.py 
365
366
367
def get_plugins_by_type(self, plugin_type: PluginType) -> list[PluginInfo]:
    """Get all plugins of a specific type"""
    return [info for info in self.plugins.values() if info.metadata.type == plugin_type]

get_status()

Get status of all plugins

Source code in tracelet/core/plugins.py 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
def get_status(self) -> dict[str, Any]:
    """Get status of all plugins"""
    status = {
        "discovered": len(self.plugins),
        "loaded": sum(1 for p in self.plugins.values() if p.state >= PluginState.LOADED),
        "active": sum(1 for p in self.plugins.values() if p.state == PluginState.ACTIVE),
        "errors": sum(1 for p in self.plugins.values() if p.state == PluginState.ERROR),
        "plugins": {},
    }

    for name, info in self.plugins.items():
        plugin_status = {
            "state": info.state.name.lower(),
            "type": info.metadata.type.value,
            "version": info.metadata.version,
        }

        if info.state == PluginState.ERROR:
            plugin_status["error"] = info.error

        if name in self._plugin_instances:
            try:
                plugin_status["status"] = self._plugin_instances[name].get_status()
            except Exception as e:
                plugin_status["status_error"] = str(e)

        status["plugins"][name] = plugin_status

    return status

initialize_plugin(plugin_name, config=None)

Initialize a plugin with configuration

Source code in tracelet/core/plugins.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
def initialize_plugin(self, plugin_name: str, config: Optional[dict[str, Any]] = None) -> bool:
    """Initialize a plugin with configuration"""
    if plugin_name not in self.plugins:
        return False

    plugin_info = self.plugins[plugin_name]

    # Ensure plugin is validated
    if plugin_info.state < PluginState.VALIDATED and not self.validate_plugin(plugin_name):
        return False

    try:
        # Create plugin instance
        plugin_class = plugin_info.instance
        instance = plugin_class()

        # Initialize with config
        instance.initialize(config or {})

        # Store instance
        self._plugin_instances[plugin_name] = instance
        plugin_info.state = PluginState.INITIALIZED

        logger.info(f"Initialized plugin: {plugin_name}")
        return True

    except Exception as e:
        plugin_info.state = PluginState.ERROR
        plugin_info.error = str(e)
        logger.exception(f"Failed to initialize plugin '{plugin_name}': {e}")
        return False

load_config(config_path)

Load plugin configuration from file

Source code in tracelet/core/plugins.py 
402
403
404
405
def load_config(self, config_path: str) -> dict[str, Any]:
    """Load plugin configuration from file"""
    with open(config_path) as f:
        return json.load(f)

load_plugin(plugin_name)

Load a specific plugin

Source code in tracelet/core/plugins.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def load_plugin(self, plugin_name: str) -> bool:
    """Load a specific plugin"""
    if plugin_name not in self.plugins:
        logger.error(f"Plugin '{plugin_name}' not found")
        return False

    plugin_info = self.plugins[plugin_name]

    if plugin_info.state >= PluginState.LOADED:
        return True

    try:
        # Load the module
        spec = importlib.util.spec_from_file_location(f"plugin_{plugin_name}", plugin_info.module_path)
        if not spec or not spec.loader:
            raise ImportError(f"Failed to load spec for {plugin_name}")

        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)

        # Get the plugin class
        plugin_class = getattr(module, plugin_info.class_name)

        # Store class reference
        plugin_info.instance = plugin_class
        plugin_info.state = PluginState.LOADED

        logger.info(f"Loaded plugin: {plugin_name}")
        return True

    except Exception as e:
        plugin_info.state = PluginState.ERROR
        plugin_info.error = str(e)
        logger.exception(f"Failed to load plugin '{plugin_name}'")
        return False

resolve_dependencies(plugin_names)

Resolve plugin dependencies and return load order

Source code in tracelet/core/plugins.py 
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
def resolve_dependencies(self, plugin_names: list[str]) -> list[str]:
    """Resolve plugin dependencies and return load order"""
    # Build dependency graph
    graph = {}
    for name in plugin_names:
        if name in self.plugins:
            deps = self.plugins[name].metadata.dependencies
            graph[name] = set(deps)

    # Topological sort
    result = []
    visited = set()
    temp_visited = set()

    def visit(node):
        if node in temp_visited:
            raise ValueError(f"Circular dependency detected: {node}")
        if node in visited:
            return

        temp_visited.add(node)
        for dep in graph.get(node, set()):
            visit(dep)
        temp_visited.remove(node)
        visited.add(node)
        result.append(node)

    for name in plugin_names:
        if name not in visited:
            visit(name)

    return result

start_plugin(plugin_name)

Start a plugin

Source code in tracelet/core/plugins.py 
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
def start_plugin(self, plugin_name: str) -> bool:
    """Start a plugin"""
    if plugin_name not in self._plugin_instances:
        logger.error(f"Plugin '{plugin_name}' not initialized")
        return False

    plugin_info = self.plugins[plugin_name]
    instance = self._plugin_instances[plugin_name]

    try:
        instance.start()
        plugin_info.state = PluginState.ACTIVE
        logger.info(f"Started plugin: {plugin_name}")
        return True

    except Exception as e:
        plugin_info.state = PluginState.ERROR
        plugin_info.error = str(e)
        logger.exception(f"Failed to start plugin '{plugin_name}': {e}")
        return False

stop_plugin(plugin_name)

Stop a plugin

Source code in tracelet/core/plugins.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
def stop_plugin(self, plugin_name: str) -> bool:
    """Stop a plugin"""
    if plugin_name not in self._plugin_instances:
        return True

    plugin_info = self.plugins[plugin_name]
    instance = self._plugin_instances[plugin_name]

    try:
        instance.stop()
        plugin_info.state = PluginState.STOPPED
        logger.info(f"Stopped plugin: {plugin_name}")
        return True

    except Exception as e:
        logger.exception(f"Error stopping plugin '{plugin_name}': {e}")
        return False

validate_plugin(plugin_name)

Validate a plugin

Source code in tracelet/core/plugins.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def validate_plugin(self, plugin_name: str) -> bool:
    """Validate a plugin"""
    if plugin_name not in self.plugins:
        return False

    plugin_info = self.plugins[plugin_name]

    if plugin_info.state < PluginState.LOADED and not self.load_plugin(plugin_name):
        return False

    try:
        # Check required methods
        required_methods = ["get_metadata", "initialize", "start", "stop", "get_status"]
        plugin_class = plugin_info.instance

        for method in required_methods:
            if not hasattr(plugin_class, method):
                raise ValueError(f"Missing required method: {method}")

        # Validate metadata
        metadata = plugin_class.get_metadata()
        if not metadata.name or not metadata.version:
            raise ValueError("Invalid metadata: name and version required")

        # Check dependencies
        for dep in metadata.dependencies:
            if dep not in self.plugins:
                raise ValueError(f"Missing dependency: {dep}")

        plugin_info.state = PluginState.VALIDATED
        logger.info(f"Validated plugin: {plugin_name}")
        return True

    except Exception as e:
        plugin_info.state = PluginState.ERROR
        plugin_info.error = str(e)
        logger.exception(f"Plugin validation failed for '{plugin_name}': {e}")
        return False

PluginMetadata(name, version, type, description='', author='', dependencies=list(), config_schema=None, capabilities=set()) dataclass

Metadata for a plugin

PluginState

Bases: IntEnum

Plugin lifecycle states

PluginType

Bases: Enum

Types of plugins supported by the system

options: show_source: true show_bases: true heading_level: 2

Overview

Tracelet's plugin system provides a flexible architecture for extending functionality through modular components. The system supports backend plugins, collector plugins, and framework integrations.

Core Plugin Architecture

Plugin Base Classes

PluginBase

Bases: ABC

Base class for all plugins

get_metadata() abstractmethod classmethod

Return plugin metadata

Source code in tracelet/core/plugins.py 
69
70
71
72
73
@classmethod
@abstractmethod
def get_metadata(cls) -> PluginMetadata:
    """Return plugin metadata"""
    pass

get_status() abstractmethod

Get plugin status

Source code in tracelet/core/plugins.py 
90
91
92
93
@abstractmethod
def get_status(self) -> dict[str, Any]:
    """Get plugin status"""
    pass

initialize(config) abstractmethod

Initialize the plugin with configuration

Source code in tracelet/core/plugins.py 
75
76
77
78
@abstractmethod
def initialize(self, config: dict[str, Any]):
    """Initialize the plugin with configuration"""
    pass

start() abstractmethod

Start the plugin

Source code in tracelet/core/plugins.py 
80
81
82
83
@abstractmethod
def start(self):
    """Start the plugin"""
    pass

stop() abstractmethod

Stop the plugin

Source code in tracelet/core/plugins.py 
85
86
87
88
@abstractmethod
def stop(self):
    """Stop the plugin"""
    pass

options: show_source: true heading_level: 4

BackendPlugin

Bases: PluginBase, BackendInterface, MetricSink

Base class for backend plugins

Source code in tracelet/core/plugins.py 
 99
100
101
def __init__(self):
    self._config = {}
    self._active = False

can_handle_type(metric_type)

Check if this backend can handle the metric type

Source code in tracelet/core/plugins.py 
107
108
109
110
def can_handle_type(self, metric_type: MetricType) -> bool:
    """Check if this backend can handle the metric type"""
    # By default, backends handle all metric types
    return True

get_sink_id()

Return sink ID for orchestrator integration

Source code in tracelet/core/plugins.py 
103
104
105
def get_sink_id(self) -> str:
    """Return sink ID for orchestrator integration"""
    return f"backend_{self.get_metadata().name}"

options: show_source: true heading_level: 4

CollectorPlugin

Bases: PluginBase, CollectorInterface

Base class for collector plugins

Source code in tracelet/core/plugins.py 
116
117
118
119
def __init__(self):
    self._config = {}
    self._active = False
    self._collection_interval = 60  # seconds

options: show_source: true heading_level: 4

Plugin Metadata

Metadata for a plugin

options: show_source: true heading_level: 3

Example Usage:

from tracelet.core.plugins import PluginMetadata

# Define plugin metadata
metadata = PluginMetadata(
    name="custom_backend",
    version="1.0.0",
    description="Custom experiment tracking backend",
    author="ML Team",
    plugin_type="backend",
    entry_point="my_package.backends.CustomBackend",
    dependencies=["requests>=2.25.0", "pandas>=1.3.0"],
    config_schema={
        "api_endpoint": {"type": "string", "required": True},
        "api_key": {"type": "string", "required": True},
        "timeout": {"type": "number", "default": 30}
    }
)

Plugin Manager

Manages plugin discovery, loading, and lifecycle

Source code in tracelet/core/plugins.py 
125
126
127
128
129
130
131
132
133
def __init__(self, plugin_paths: Optional[list[str]] = None, use_default_paths: bool = True):
    self.plugin_paths = plugin_paths or []
    self.plugins: dict[str, PluginInfo] = {}
    self._plugin_instances: dict[str, PluginBase] = {}
    self._dependency_graph: dict[str, set[str]] = {}

    # Add default plugin paths unless explicitly disabled
    if use_default_paths:
        self._add_default_paths()

discover_plugins()

Discover all available plugins

Source code in tracelet/core/plugins.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def discover_plugins(self) -> list[PluginInfo]:
    """Discover all available plugins"""
    discovered = []

    for path in self.plugin_paths:
        if os.path.isdir(path):
            discovered.extend(self._discover_in_directory(path))
        elif path.endswith(".py"):
            plugin = self._discover_in_file(path)
            if plugin:
                discovered.append(plugin)

    # Update internal registry
    for plugin in discovered:
        self.plugins[plugin.metadata.name] = plugin
        logger.info(f"Discovered plugin: {plugin.metadata.name} v{plugin.metadata.version}")

    return discovered

get_plugin_instance(plugin_name)

Get an initialized plugin instance

Source code in tracelet/core/plugins.py 
361
362
363
def get_plugin_instance(self, plugin_name: str) -> Optional[PluginBase]:
    """Get an initialized plugin instance"""
    return self._plugin_instances.get(plugin_name)

get_plugins_by_type(plugin_type)

Get all plugins of a specific type

Source code in tracelet/core/plugins.py 
365
366
367
def get_plugins_by_type(self, plugin_type: PluginType) -> list[PluginInfo]:
    """Get all plugins of a specific type"""
    return [info for info in self.plugins.values() if info.metadata.type == plugin_type]

get_status()

Get status of all plugins

Source code in tracelet/core/plugins.py 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
def get_status(self) -> dict[str, Any]:
    """Get status of all plugins"""
    status = {
        "discovered": len(self.plugins),
        "loaded": sum(1 for p in self.plugins.values() if p.state >= PluginState.LOADED),
        "active": sum(1 for p in self.plugins.values() if p.state == PluginState.ACTIVE),
        "errors": sum(1 for p in self.plugins.values() if p.state == PluginState.ERROR),
        "plugins": {},
    }

    for name, info in self.plugins.items():
        plugin_status = {
            "state": info.state.name.lower(),
            "type": info.metadata.type.value,
            "version": info.metadata.version,
        }

        if info.state == PluginState.ERROR:
            plugin_status["error"] = info.error

        if name in self._plugin_instances:
            try:
                plugin_status["status"] = self._plugin_instances[name].get_status()
            except Exception as e:
                plugin_status["status_error"] = str(e)

        status["plugins"][name] = plugin_status

    return status

initialize_plugin(plugin_name, config=None)

Initialize a plugin with configuration

Source code in tracelet/core/plugins.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
def initialize_plugin(self, plugin_name: str, config: Optional[dict[str, Any]] = None) -> bool:
    """Initialize a plugin with configuration"""
    if plugin_name not in self.plugins:
        return False

    plugin_info = self.plugins[plugin_name]

    # Ensure plugin is validated
    if plugin_info.state < PluginState.VALIDATED and not self.validate_plugin(plugin_name):
        return False

    try:
        # Create plugin instance
        plugin_class = plugin_info.instance
        instance = plugin_class()

        # Initialize with config
        instance.initialize(config or {})

        # Store instance
        self._plugin_instances[plugin_name] = instance
        plugin_info.state = PluginState.INITIALIZED

        logger.info(f"Initialized plugin: {plugin_name}")
        return True

    except Exception as e:
        plugin_info.state = PluginState.ERROR
        plugin_info.error = str(e)
        logger.exception(f"Failed to initialize plugin '{plugin_name}': {e}")
        return False

load_config(config_path)

Load plugin configuration from file

Source code in tracelet/core/plugins.py 
402
403
404
405
def load_config(self, config_path: str) -> dict[str, Any]:
    """Load plugin configuration from file"""
    with open(config_path) as f:
        return json.load(f)

load_plugin(plugin_name)

Load a specific plugin

Source code in tracelet/core/plugins.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def load_plugin(self, plugin_name: str) -> bool:
    """Load a specific plugin"""
    if plugin_name not in self.plugins:
        logger.error(f"Plugin '{plugin_name}' not found")
        return False

    plugin_info = self.plugins[plugin_name]

    if plugin_info.state >= PluginState.LOADED:
        return True

    try:
        # Load the module
        spec = importlib.util.spec_from_file_location(f"plugin_{plugin_name}", plugin_info.module_path)
        if not spec or not spec.loader:
            raise ImportError(f"Failed to load spec for {plugin_name}")

        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)

        # Get the plugin class
        plugin_class = getattr(module, plugin_info.class_name)

        # Store class reference
        plugin_info.instance = plugin_class
        plugin_info.state = PluginState.LOADED

        logger.info(f"Loaded plugin: {plugin_name}")
        return True

    except Exception as e:
        plugin_info.state = PluginState.ERROR
        plugin_info.error = str(e)
        logger.exception(f"Failed to load plugin '{plugin_name}'")
        return False

resolve_dependencies(plugin_names)

Resolve plugin dependencies and return load order

Source code in tracelet/core/plugins.py 
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
def resolve_dependencies(self, plugin_names: list[str]) -> list[str]:
    """Resolve plugin dependencies and return load order"""
    # Build dependency graph
    graph = {}
    for name in plugin_names:
        if name in self.plugins:
            deps = self.plugins[name].metadata.dependencies
            graph[name] = set(deps)

    # Topological sort
    result = []
    visited = set()
    temp_visited = set()

    def visit(node):
        if node in temp_visited:
            raise ValueError(f"Circular dependency detected: {node}")
        if node in visited:
            return

        temp_visited.add(node)
        for dep in graph.get(node, set()):
            visit(dep)
        temp_visited.remove(node)
        visited.add(node)
        result.append(node)

    for name in plugin_names:
        if name not in visited:
            visit(name)

    return result

start_plugin(plugin_name)

Start a plugin

Source code in tracelet/core/plugins.py 
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
def start_plugin(self, plugin_name: str) -> bool:
    """Start a plugin"""
    if plugin_name not in self._plugin_instances:
        logger.error(f"Plugin '{plugin_name}' not initialized")
        return False

    plugin_info = self.plugins[plugin_name]
    instance = self._plugin_instances[plugin_name]

    try:
        instance.start()
        plugin_info.state = PluginState.ACTIVE
        logger.info(f"Started plugin: {plugin_name}")
        return True

    except Exception as e:
        plugin_info.state = PluginState.ERROR
        plugin_info.error = str(e)
        logger.exception(f"Failed to start plugin '{plugin_name}': {e}")
        return False

stop_plugin(plugin_name)

Stop a plugin

Source code in tracelet/core/plugins.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
def stop_plugin(self, plugin_name: str) -> bool:
    """Stop a plugin"""
    if plugin_name not in self._plugin_instances:
        return True

    plugin_info = self.plugins[plugin_name]
    instance = self._plugin_instances[plugin_name]

    try:
        instance.stop()
        plugin_info.state = PluginState.STOPPED
        logger.info(f"Stopped plugin: {plugin_name}")
        return True

    except Exception as e:
        logger.exception(f"Error stopping plugin '{plugin_name}': {e}")
        return False

validate_plugin(plugin_name)

Validate a plugin

Source code in tracelet/core/plugins.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def validate_plugin(self, plugin_name: str) -> bool:
    """Validate a plugin"""
    if plugin_name not in self.plugins:
        return False

    plugin_info = self.plugins[plugin_name]

    if plugin_info.state < PluginState.LOADED and not self.load_plugin(plugin_name):
        return False

    try:
        # Check required methods
        required_methods = ["get_metadata", "initialize", "start", "stop", "get_status"]
        plugin_class = plugin_info.instance

        for method in required_methods:
            if not hasattr(plugin_class, method):
                raise ValueError(f"Missing required method: {method}")

        # Validate metadata
        metadata = plugin_class.get_metadata()
        if not metadata.name or not metadata.version:
            raise ValueError("Invalid metadata: name and version required")

        # Check dependencies
        for dep in metadata.dependencies:
            if dep not in self.plugins:
                raise ValueError(f"Missing dependency: {dep}")

        plugin_info.state = PluginState.VALIDATED
        logger.info(f"Validated plugin: {plugin_name}")
        return True

    except Exception as e:
        plugin_info.state = PluginState.ERROR
        plugin_info.error = str(e)
        logger.exception(f"Plugin validation failed for '{plugin_name}': {e}")
        return False

options: show_source: true heading_level: 3

Plugin Discovery and Registration

from tracelet.core.plugins import PluginManager

# Initialize plugin manager
plugin_manager = PluginManager()

# Discover installed plugins
plugin_manager.discover_plugins()

# List available plugins
backend_plugins = plugin_manager.list_plugins("backend")
collector_plugins = plugin_manager.list_plugins("collector")

print(f"Available backend plugins: {[p.name for p in backend_plugins]}")
print(f"Available collector plugins: {[p.name for p in collector_plugins]}")

Manual Plugin Registration

from my_package.plugins import CustomBackendPlugin

# Register plugin manually
plugin_manager.register_plugin(CustomBackendPlugin)

# Get specific plugin
plugin_class = plugin_manager.get_plugin("backend", "custom_backend")
if plugin_class:
    plugin_instance = plugin_class()

Creating Custom Plugins

Backend Plugin Example

from tracelet.core.plugins import BackendPlugin, PluginMetadata
from typing import Dict, Any, Optional

class CustomBackendPlugin(BackendPlugin):
    """Custom backend for proprietary experiment tracking system."""

    def get_metadata(self) -> PluginMetadata:
        return PluginMetadata(
            name="custom_backend",
            version="1.0.0",
            description="Integration with Custom Tracking System",
            author="Your Team",
            plugin_type="backend",
            entry_point="custom_package.CustomBackendPlugin",
            dependencies=["custom-sdk>=2.0.0"],
            config_schema={
                "server_url": {"type": "string", "required": True},
                "project_id": {"type": "string", "required": True},
                "api_token": {"type": "string", "required": True}
            }
        )

    def initialize(self, config: Dict[str, Any]) -> None:
        """Initialize the backend with configuration."""
        self.server_url = config["server_url"]
        self.project_id = config["project_id"]
        self.api_token = config["api_token"]

        # Initialize SDK
        import custom_sdk
        self.client = custom_sdk.Client(
            url=self.server_url,
            token=self.api_token
        )

        # Create or get project
        self.project = self.client.get_project(self.project_id)
        self.experiment = None

    def start(self) -> None:
        """Start experiment tracking."""
        self.experiment = self.project.create_experiment(
            name=self.config.get("experiment_name", "Default Experiment")
        )
        self.experiment.start()

    def stop(self) -> None:
        """Stop experiment tracking."""
        if self.experiment:
            self.experiment.finish()
            self.experiment = None

    def log_metric(self, name: str, value: float, iteration: Optional[int] = None) -> None:
        """Log a metric value."""
        if self.experiment:
            self.experiment.log_metric(
                name=name,
                value=value,
                step=iteration
            )

    def log_params(self, params: Dict[str, Any]) -> None:
        """Log parameters."""
        if self.experiment:
            self.experiment.log_params(params)

    def log_artifact(self, local_path: str, artifact_path: str) -> None:
        """Log an artifact."""
        if self.experiment:
            self.experiment.upload_artifact(local_path, artifact_path)

    def get_status(self) -> Dict[str, Any]:
        """Get backend status."""
        return {
            "backend_name": "custom_backend",
            "connected": self.experiment is not None,
            "experiment_id": getattr(self.experiment, "id", None),
            "project_id": self.project_id
        }

Collector Plugin Example

from tracelet.core.plugins import CollectorPlugin, PluginMetadata
import psutil
import time
from typing import Dict, Any

class AdvancedSystemCollectorPlugin(CollectorPlugin):
    """Advanced system metrics collector with custom metrics."""

    def get_metadata(self) -> PluginMetadata:
        return PluginMetadata(
            name="advanced_system_collector",
            version="1.2.0",
            description="Advanced system metrics with process-level details",
            author="System Team",
            plugin_type="collector",
            entry_point="system_package.AdvancedSystemCollectorPlugin",
            dependencies=["psutil>=5.8.0", "GPUtil>=1.4.0"],
            config_schema={
                "include_processes": {"type": "boolean", "default": False},
                "process_limit": {"type": "number", "default": 10},
                "include_network": {"type": "boolean", "default": True}
            }
        )

    def initialize(self, config: Dict[str, Any]) -> None:
        """Initialize collector with configuration."""
        self.include_processes = config.get("include_processes", False)
        self.process_limit = config.get("process_limit", 10)
        self.include_network = config.get("include_network", True)
        self.last_network = None

    def start(self) -> None:
        """Start collection (if background collection needed)."""
        self.start_time = time.time()
        if self.include_network:
            self.last_network = psutil.net_io_counters()

    def stop(self) -> None:
        """Stop collection."""
        pass

    def collect(self) -> Dict[str, Any]:
        """Collect system metrics."""
        metrics = {
            "timestamp": time.time(),
            "uptime": time.time() - self.start_time
        }

        # CPU metrics
        cpu_percent = psutil.cpu_percent(interval=1)
        cpu_count = psutil.cpu_count()
        metrics.update({
            "cpu_percent": cpu_percent,
            "cpu_count_logical": cpu_count,
            "cpu_count_physical": psutil.cpu_count(logical=False),
            "cpu_freq": psutil.cpu_freq()._asdict() if psutil.cpu_freq() else {}
        })

        # Memory metrics
        memory = psutil.virtual_memory()
        swap = psutil.swap_memory()
        metrics.update({
            "memory_total": memory.total,
            "memory_available": memory.available,
            "memory_percent": memory.percent,
            "swap_total": swap.total,
            "swap_used": swap.used,
            "swap_percent": swap.percent
        })

        # Disk metrics
        disk_usage = psutil.disk_usage('/')
        disk_io = psutil.disk_io_counters()
        metrics.update({
            "disk_total": disk_usage.total,
            "disk_used": disk_usage.used,
            "disk_percent": disk_usage.percent,
            "disk_read_bytes": disk_io.read_bytes if disk_io else 0,
            "disk_write_bytes": disk_io.write_bytes if disk_io else 0
        })

        # Network metrics (with rate calculation)
        if self.include_network:
            current_network = psutil.net_io_counters()
            if self.last_network:
                time_delta = 1.0  # Approximate
                metrics.update({
                    "network_bytes_sent": current_network.bytes_sent,
                    "network_bytes_recv": current_network.bytes_recv,
                    "network_bytes_sent_rate": (current_network.bytes_sent - self.last_network.bytes_sent) / time_delta,
                    "network_bytes_recv_rate": (current_network.bytes_recv - self.last_network.bytes_recv) / time_delta
                })
            self.last_network = current_network

        # Process-level metrics
        if self.include_processes:
            processes = []
            for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
                try:
                    processes.append(proc.info)
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    pass

            # Sort by CPU usage and take top N
            processes.sort(key=lambda x: x['cpu_percent'] or 0, reverse=True)
            metrics["top_processes"] = processes[:self.process_limit]

        # GPU metrics (if available)
        try:
            import GPUtil
            gpus = GPUtil.getGPUs()
            gpu_metrics = []
            for gpu in gpus:
                gpu_metrics.append({
                    "id": gpu.id,
                    "name": gpu.name,
                    "load": gpu.load * 100,
                    "memory_util": gpu.memoryUtil * 100,
                    "temperature": gpu.temperature
                })
            metrics["gpu"] = gpu_metrics
        except ImportError:
            pass

        return metrics

    def get_status(self) -> Dict[str, Any]:
        """Get collector status."""
        return {
            "collector_name": "advanced_system_collector",
            "active": True,
            "config": {
                "include_processes": self.include_processes,
                "include_network": self.include_network
            }
        }

Plugin Installation and Packaging

setuptools Entry Points

# setup.py for your plugin package
from setuptools import setup, find_packages

setup(
    name="tracelet-custom-plugins",
    version="1.0.0",
    packages=find_packages(),
    install_requires=[
        "tracelet>=1.0.0",
        "psutil>=5.8.0"
    ],
    entry_points={
        "tracelet.plugins": [
            "custom_backend = my_package.plugins:CustomBackendPlugin",
            "advanced_system = my_package.collectors:AdvancedSystemCollectorPlugin"
        ]
    }
)

Plugin Discovery

# Tracelet automatically discovers plugins via entry points
import pkg_resources

def discover_plugins():
    """Discover installed Tracelet plugins."""
    plugins = {}

    for entry_point in pkg_resources.iter_entry_points("tracelet.plugins"):
        try:
            plugin_class = entry_point.load()
            plugin_instance = plugin_class()
            metadata = plugin_instance.get_metadata()
            plugins[metadata.name] = {
                "class": plugin_class,
                "metadata": metadata,
                "entry_point": entry_point
            }
        except Exception as e:
            print(f"Failed to load plugin {entry_point.name}: {e}")

    return plugins

Advanced Plugin Features

Plugin Dependencies

from tracelet.core.plugins import PluginManager
import importlib

def check_plugin_dependencies(plugin_metadata):
    """Check if plugin dependencies are satisfied."""
    missing_deps = []

    for dep in plugin_metadata.dependencies:
        try:
            # Parse dependency specification (name>=version)
            if ">=" in dep:
                package_name = dep.split(">=")[0]
            else:
                package_name = dep

            importlib.import_module(package_name)
        except ImportError:
            missing_deps.append(dep)

    return missing_deps

# Usage
plugin_manager = PluginManager()
plugins = plugin_manager.discover_plugins()

for plugin_name, plugin_info in plugins.items():
    missing = check_plugin_dependencies(plugin_info["metadata"])
    if missing:
        print(f"Plugin {plugin_name} missing dependencies: {missing}")

Configuration Validation

import jsonschema
from typing import Dict, Any

def validate_plugin_config(config: Dict[str, Any], schema: Dict[str, Any]) -> bool:
    """Validate plugin configuration against schema."""
    try:
        jsonschema.validate(config, schema)
        return True
    except jsonschema.ValidationError as e:
        print(f"Configuration validation failed: {e}")
        return False

# Example usage in plugin
class ValidatedBackendPlugin(BackendPlugin):
    def initialize(self, config: Dict[str, Any]) -> None:
        metadata = self.get_metadata()

        if not validate_plugin_config(config, metadata.config_schema):
            raise ValueError("Invalid plugin configuration")

        # Proceed with initialization
        super().initialize(config)

Plugin Lifecycle Management

from tracelet.core.plugins import PluginManager
from typing import List

class PluginLifecycleManager:
    """Manages plugin lifecycle across experiment sessions."""

    def __init__(self):
        self.plugin_manager = PluginManager()
        self.active_plugins: List[PluginBase] = []

    def start_plugins(self, plugin_configs: Dict[str, Dict[str, Any]]):
        """Start configured plugins."""
        for plugin_name, config in plugin_configs.items():
            try:
                plugin_class = self.plugin_manager.get_plugin_by_name(plugin_name)
                if plugin_class:
                    plugin_instance = plugin_class()
                    plugin_instance.initialize(config)
                    plugin_instance.start()
                    self.active_plugins.append(plugin_instance)
            except Exception as e:
                print(f"Failed to start plugin {plugin_name}: {e}")

    def stop_all_plugins(self):
        """Stop all active plugins."""
        for plugin in self.active_plugins:
            try:
                plugin.stop()
            except Exception as e:
                print(f"Error stopping plugin: {e}")

        self.active_plugins.clear()

    def get_plugin_status(self) -> Dict[str, Any]:
        """Get status of all active plugins."""
        status = {}
        for plugin in self.active_plugins:
            try:
                plugin_status = plugin.get_status()
                status[plugin_status.get("name", "unknown")] = plugin_status
            except Exception as e:
                status["error"] = str(e)

        return status

Best Practices

Plugin Development

  1. Follow Interface Contracts: Implement all required methods from base classes
  2. Error Handling: Handle errors gracefully and provide meaningful error messages
  3. Configuration Validation: Validate configuration early and provide clear feedback
  4. Resource Management: Properly clean up resources in stop() method
  5. Documentation: Provide clear documentation and examples

Plugin Usage

  1. Dependency Management: Check plugin dependencies before use
  2. Configuration Security: Don't expose sensitive information in configuration
  3. Performance: Monitor plugin performance impact on main application
  4. Testing: Test plugins in isolation and integration scenarios
  5. Version Compatibility: Ensure plugin versions are compatible with Tracelet core