This guide provides comprehensive debugging techniques and workflows for troubleshooting Pyvider providers. Whether you're tracking down a subtle bug or dealing with a complete failure, these strategies will help you identify and fix issues quickly.
🤖 AI-Generated Content
This documentation was generated with AI assistance and is still being audited. Some, or potentially a lot, of this information may be inaccurate. Learn more.
# In your provider codefromprovide.foundationimportloggerlogger.debug("Detailed debugging information")# Developmentlogger.info("Important state changes")# Normal operationslogger.warning("Recoverable issues")# Potential problemslogger.error("Failures requiring attention")# Errors
# Show only ERROR logsgrep"ERROR"terraform-debug.log
# Show logs for specific resourcegrep"pyvider_file_content"terraform-debug.log
# Show specific operationgrep"operation=create"terraform-debug.log
# Show logs with timinggrep"duration"terraform-debug.log
@register_resource("pyvider_example")classExampleResource(BaseResource):asyncdef_create(self,ctx:ResourceContext,base_plan:dict):logger.debug("Create operation starting",resource_class=self.__class__.__name__,config=base_plan)try:result=awaitself.api.create_resource(base_plan)logger.debug("Resource created successfully",resource_id=result.id,duration_ms=result.duration)return{**base_plan,"id":result.id},NoneexceptExceptionase:logger.error("Failed to create resource",error=str(e),error_type=type(e).__name__,config=base_plan)raise
# Navigationn# Next lines# Step into functionc# Continue executionr# Return from current function# Inspectionpvariable_name# Print variableppvariable_name# Pretty-print variablel# List code around current linew# Show stack trace# Breakpointsbline_number# Set breakpoint at linebfunction_name# Set breakpoint at functioncl# Clear all breakpoints# Otherh# Helpq# Quit debugger
asyncdef_create(self,ctx:ResourceContext,base_plan:dict):# Set breakpoint before awaitbreakpoint()# When you hit this breakpoint, you can:# - Step over the await with 'n'# - Step into the async function with 's'result=awaitself.api.create_resource(base_plan)# Set another breakpoint to inspect resultbreakpoint()returnresult,None
asyncdefread(self,ctx:ResourceContext):filename=ctx.state.filename# Only break for specific filesiffilename=="/tmp/debug-this.txt":breakpoint()content=safe_read_text(Path(filename))returnFileContentState(...)
# At breakpointbreakpoint()# Inspect configuration(Pdb)ppctx.config.__dict__{'filename':'/tmp/test.txt','content':'Hello'}# Inspect state (if exists)(Pdb)ppctx.state.__dict__ifctx.stateelse"No state"{'filename':'/tmp/test.txt','content':'Hello','exists':True}# Check if field is unknown during planning(Pdb)print(ctx.is_field_unknown('content'))False# Check private state(Pdb)print(ctx.private_state)None
# Maximum verbosity for gRPCexportTF_LOG=TRACE
exportGRPC_GO_LOG_VERBOSITY_LEVEL=99exportGRPC_GO_LOG_SEVERITY_LEVEL=info
terraformapply2>&1|grep-i"grpc"
asyncdef_create(self,ctx:ResourceContext,base_plan:dict):logger.debug("_create called",config=base_plan)breakpoint()# Pause to inspect# ... rest of method
asyncdefread(self,ctx:ResourceContext):logger.debug("read() called",state=ctx.state.__dict__)# Should return None if resource doesn't exist# Should return updated state if changed
importsyssys.settrace(lambda*args:None)# Enable full tracebacksasyncdef_create(self,ctx:ResourceContext,base_plan:dict):breakpoint()# Will stop before crashresult=awaitself.api.create(...)returnresult,None
@classmethoddefget_schema(cls)->PvsSchema:schema=s_resource({"filename":a_str(required=True),# Must match Terraform"content":a_str(required=True),})logger.debug("Schema defined",attributes=list(schema.attributes.keys()))returnschema
# In _create, log what you receivedasyncdef_create(self,ctx:ResourceContext,base_plan:dict):logger.debug("Received attributes",attributes=list(base_plan.keys()))# Should match schema exactly
# Show entire stateterraformshow
# Show specific resourceterraformshow-json|jq'.values.root_module.resources[] | select(.address=="pyvider_file_content.test")'# Show state file directlycatterraform.tfstate|jq'.resources[]'
asyncdef_update(self,ctx:ResourceContext,base_plan:dict):logger.debug("State transition",old_state=ctx.state.__dict__ifctx.stateelseNone,new_plan=base_plan,differences=self._compute_diff(ctx.state,base_plan))result=awaitself.api.update(...)returnresult,Nonedef_compute_diff(self,old,new):"""Helper to show what changed."""ifnotold:return"creating"changes={}forkeyinnew:ifgetattr(old,key,None)!=new[key]:changes[key]={"old":getattr(old,key),"new":new[key]}returnchanges
frompyvider.resourcesimportPrivateStateasyncdefread(self,ctx:ResourceContext):# Decrypt and inspect private stateifctx.private_state:try:decrypted=PrivateState.decrypt(ctx.private_state)logger.debug("Private state contents",data=decrypted)exceptExceptionase:logger.error("Failed to decrypt private state",error=str(e))returncurrent_state
asyncdef_create(self,ctx:ResourceContext,base_plan:dict):result=awaitself.api.create(...)# Return state must include ALL attributes from schemastate={**base_plan,# Include all config"id":result.id,"exists":True,# Computed attribute"content_hash":self._compute_hash(base_plan["content"]),# Computed}# Log for verificationlogger.debug("Returning state",state_keys=list(state.keys()),schema_attrs=list(self.get_schema().attributes.keys()))returnstate,None
@classmethoddefget_schema(cls)->PvsSchema:schema=s_resource({"filename":a_str(required=True,description="Path to the file",validators=[lambdax:notx.startswith("/")or"Use relative paths"]),"content":a_str(required=True),"exists":a_bool(computed=True),})# Debug: Print schemalogger.debug("Schema created",attributes={k:{"required":v.required,"computed":v.computed}fork,vinschema.attributes.items()})returnschema
# Test required attributesresource"pyvider_file_content""test_required"{ # Missing required attribute - should errorfilename="/tmp/test.txt" # content missing!}# Test computed attributesresource"pyvider_file_content""test_computed"{filename="/tmp/test.txt"content="test" # Don't set computed attributes # exists = true # This should error}
asyncdef_create(self,ctx:ResourceContext,base_plan:dict):# Log types receivedlogger.debug("Received types",types={k:type(v).__name__fork,vinbase_plan.items()})# Verify types match schemaifnotisinstance(base_plan.get("filename"),str):logger.error("filename is not string!",type=type(base_plan["filename"]))returnbase_plan,None
importtimefromfunctoolsimportwrapsdeftimed(operation_name):"""Decorator to time operations."""defdecorator(func):@wraps(func)asyncdefwrapper(*args,**kwargs):start=time.time()try:result=awaitfunc(*args,**kwargs)duration=time.time()-startlogger.debug(f"{operation_name} completed",duration_ms=int(duration*1000))returnresultexceptExceptionase:duration=time.time()-startlogger.error(f"{operation_name} failed",duration_ms=int(duration*1000),error=str(e))raisereturnwrapperreturndecorator@register_resource("pyvider_example")classExampleResource(BaseResource):@timed("read")asyncdefread(self,ctx:ResourceContext):returnawaitself._fetch_state()@timed("create")asyncdef_create(self,ctx:ResourceContext,base_plan:dict):returnawaitself._do_create(base_plan)
# Real-time profilingpy-spytop--pid$(pgrep-fterraform-provider-pyvider)# Generate flame graphpy-spyrecord-oprofile.svg--pid$(pgrep-fterraform-provider-pyvider)# Record for specific durationpy-spyrecord-oprofile.svg--duration60--pid<pid>
importtracemallocclassMyProvider(BaseProvider):asyncdefconfigure(self,config:ProviderConfig):# Start trackingtracemalloc.start()asyncdefcleanup(self):# Get memory statisticssnapshot=tracemalloc.take_snapshot()top_stats=snapshot.statistics('lineno')logger.debug("Memory usage top 10:")forstatintop_stats[:10]:logger.debug(f"{stat}")
# Using mitmproxymitmproxy--modetransparent
# Or charles proxycharles-proxy
# Configure provider to use proxyexportHTTP_PROXY=http://localhost:8080
exportHTTPS_PROXY=http://localhost:8080
# Test API calls outside Terraformimportasyncioimporthttpxasyncdeftest_api():asyncwithhttpx.AsyncClient()asclient:response=awaitclient.post("https://api.example.com/resource",json={"name":"test"})print(response.status_code)print(response.json())asyncio.run(test_api())
# Run with debuggerpytesttests/test_my_resource.py--pdb
# Stop at first failurepytesttests/-x--pdb
# Verbose outputpytesttests/-vv
# Show print statementspytesttests/-s
@pytest.fixturedefdebug_ctx(capsys):"""Fixture that prints context for debugging."""def_debug(ctx):print(f"\nContext: {ctx.__dict__}")print(f"Config: {ctx.config.__dict__ifctx.configelseNone}")print(f"State: {ctx.state.__dict__ifctx.stateelseNone}")return_debugasyncdeftest_with_debug(debug_ctx):ctx=make_context(...)debug_ctx(ctx)# Print for inspectionresult=awaitresource.read(ctx)assertresultisnotNone
importthreadingdefdebug_threads():"""Print all active threads."""forthreadinthreading.enumerate():logger.debug("Active thread",name=thread.name,daemon=thread.daemon,alive=thread.is_alive())# Call when debugging hangsdebug_threads()
Remember: Debugging is a skill that improves with practice. Start with simple logging, escalate to interactive debugging when needed, and always work systematically from symptoms to root cause.