I had an operator that stores a dataframe to a bigquery tabl

Question

I had an operator that stores a dataframe to a bigquery table and when I run locally i believe the code automatically pulls my credential file in my local home directory to authenticate how do secrets work with dagster cloud in the local agent configuration I tested out the same operator in the cloud run and it just worked on the first try

daniel · Answer

Since the agent runs your ops I d expect that to work the same way as in open source dagster dagster cloud doesn t need any additional access to your secrets in order to work

Rohan Kshirsagar · Answer

so if I terminated my local agent the jobs synced to dagster cloud stop working

Rohan Kshirsagar · Answer

oh i see the local agent is basically running everything locally on my machine so I can t actually shut my laptop down and expect my dagster cloud pipelines to run

Rohan Kshirsagar · Answer

so whats the difference between local dagster with dagit and dagster cloud with local agent

Rohan Kshirsagar · Answer

sorry for all the questions im trying to figure out the fastest path how to schedule my data ml pipelines on the cloud with cloud compute sweat smile

daniel · Answer

That s right cloud handles scheduling orchestration storage UI and the parts that require access to your code data happen via the agent

daniel · Answer

No problem One big difference between local dagster with dagit is that you ll never have to worry about anything related to storage or database migrations or hosting dagit There s also built in SSO and RBAC features

Rohan Kshirsagar · Answer

okay I see thanks thats helpful and a pretty cool design so I just need to figure out how to figure out the simplest way to run an agent with some cloud compute I was hoping for a more lazy compute option that only spun up compute on scheduled runs

daniel · Answer

If you want your pipelines to run with cloud compute your best bet is probably to use our kubernetes or ECS agents <https docs dagster cloud deployment the agent|https docs dagster cloud deployment the agent>

daniel · Answer

Running the local agent somewhere like EC2 would also be an option depending on your compute needs

Rohan Kshirsagar · Answer

I ll probably work on that next week using google cloud perhaps we can work together to generate some docs on that

daniel · Answer

that would be great We definitely could improve the documentation around best use of the local agent

Rohan Kshirsagar · Answer

Getting back to this finally If I m on Google Cloud which offering would be the right set up to run the local agent Is it Google Compute Engine Wondering if you ve seen any examples of this integration

daniel · Answer

GCE is probably the way to go If you re planning to run in the cloud I d consider trying the Docker Agent <https docs dagster cloud agents docker setup> That has a somewhat better CI CD story than the local agent For examples we re working on a Cloudformation template for the Docker agent that should be ready sometime this week That s AWS not Google Cloud but could still function as an example for getting cloud infra up and running

Rohan Kshirsagar · Answer

i was able to get it up and running and the jobs are starting up fine but failing in the middle i wonder if there are some permissions firewall issues with gprc ```The above exception was caused by the following exception grpc channel MultiThreadedRendezvous lt MultiThreadedRendezvous of RPC that terminated with status = StatusCode UNAVAILABLE details = failed to connect to all addresses debug error string = created 1646066017 618437961 description Failed to pick subchannel file src core ext filters client channel client channel cc file line 3128 referenced errors created 1646066017 618436560 description failed to connect to all addresses file src core lib transport error utils cc file line 163 grpc status 14 ```

daniel · Answer

Would it be possible to post the full stack trace

Rohan Kshirsagar · Answer

```2022 02 28 16 33 25 +0000 dagster ERROR topic job 7596ba94 1f22 4222 8285 49aad84705a9 2757 assign topics STEP FAILURE Execution of step assign topics failed dagster core executor child process executor ChildProcessCrashException Stack Trace File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster core executor multiprocess py line 210 in execute event or none = next step iter File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster core executor multiprocess py line 323 in execute step out of process for ret in execute child process command multiproc ctx command File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster core executor child process executor py line 158 in execute child process command raise ChildProcessCrashException exit code=process exitcode 2022 02 28 16 33 26 +0000 dagster ERROR topic job 7596ba94 1f22 4222 8285 49aad84705a9 store to table Dependencies for step store to table failed assign topics Not executing ```

Rohan Kshirsagar · Answer

```Stack Trace File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud agent dagster cloud agent py line 541 in process api request api result = self handle api request request instance user code launcher File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud agent dagster cloud agent py line 390 in handle api request serialized sensor data or error = client external sensor execution File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster grpc client py line 292 in external sensor execution chunks = list File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster grpc client py line 117 in streaming query raise DagsterUserCodeUnreachableError Could not reach user code server from e The above exception was caused by the following exception grpc channel MultiThreadedRendezvous lt MultiThreadedRendezvous of RPC that terminated with status = StatusCode UNAVAILABLE details = failed to connect to all addresses debug error string = created 1646066137 749912668 description Failed to pick subchannel file src core ext filters client channel client channel cc file line 3128 referenced errors created 1646066137 749911537 description failed to connect to all addresses file src core lib transport error utils cc file line 163 grpc status 14 gt Stack Trace File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster grpc client py line 115 in streaming query yield from response stream File home rohan miniconda3 envs unify news dag lib python3 8 site packages grpc channel py line 426 in next return self next File home rohan miniconda3 envs unify news dag lib python3 8 site packages grpc channel py line 809 in next raise self```

daniel · Answer

Got it I think those are two different errors Where did you see the second one

Rohan Kshirsagar · Answer

gotcha yeah now it erroring with the gprc error every few seconds I tried restarting the cloud agent im not sure if thats a function of the cloud server trying to re execute the job ```2022 02 28 16 38 41 +0000 dagster cloud ERROR dagster core errors DagsterUserCodeUnreachableError Failure loading server endpoint for location unify news dag Exception Timed out waiting for gRPC server to start with arguments home rohan miniconda3 envs unify news dag bin python m dagster api grpc lazy load user code socket tmp tmp07zmf4q heartbeat heartbeat timeout 60 log level WARNING use python environment entry point d home rohan news bias package name unify news dag Most recent connection error dagster core errors DagsterUserCodeUnreachableError Could not reach user code server Stack Trace File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster grpc server py line 933 in wait for grpc server client ping File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster grpc client py line 121 in ping res = self query Ping api pb2 PingRequest echo=echo File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster grpc client py line 108 in query raise DagsterUserCodeUnreachableError Could not reach user code server from e The above exception was caused by the following exception grpc channel InactiveRpcError lt InactiveRpcError of RPC that terminated with status = StatusCode UNAVAILABLE details = failed to connect to all addresses debug error string = created 1646066257 267542960 description Failed to pick subchannel file src core ext filters client channel client channel cc file line 3128 referenced errors created 1646066257 267541769 description failed to connect to all addresses file src core lib transport error utils cc file line 163 grpc status 14 gt Stack Trace File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster grpc client py line 105 in query response = getattr stub method request type kwargs timeout=timeout File home rohan miniconda3 envs unify news dag lib python3 8 site packages grpc channel py line 946 in call return end unary response blocking state call False None File home rohan miniconda3 envs unify news dag lib python3 8 site packages grpc channel py line 849 in end unary response blocking raise InactiveRpcError state Stack Trace File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud workspace user code launcher user code launcher py line 410 in reconcile new updated endpoint = self create new server endpoint File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud workspace user code launcher process py line 128 in create new server endpoint server process = GrpcServerProcess File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster grpc server py line 1091 in init self server process = open server process File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster grpc server py line 1004 in open server process wait for grpc server server process client subprocess args timeout=startup timeout File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster grpc server py line 939 in wait for grpc server raise Exception Stack Trace File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud agent dagster cloud agent py line 541 in process api request api result = self handle api request request instance user code launcher File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud agent dagster cloud agent py line 384 in handle api request client = self get grpc client File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud agent dagster cloud agent py line 269 in get grpc client endpoint = user code launcher get grpc endpoint repository location origin File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud workspace user code launcher user code launcher py line 511 in get grpc endpoint raise DagsterUserCodeUnreachableError ```

daniel · Answer

the ChildProcessCrashException happened because the subprocess that it opened to run your op crashed The most frequent reason i ve seen for this happening is because of a memory issue in the subprocess causing it to run out of memory You might get more useful output of the error by running the job using the in process executor by adding this to your run config in the launchpad That runs all the ops in the same process ```execution config in process ```

daniel · Answer

I think overall the symptoms that you re seeing are consistent with your agent and jobs using more resources than you have available on your GCE box i e the operating system seems to be shutting down a bunch of your processes One of the nice things about using the docker agent is that it gets easier to enforce resource limits in the different processes that the agent spins up

daniel · Answer

These warnings from the docs about the local agent may apply here <https docs dagster cloud agents local setup considerations when running the local agent in production>

Rohan Kshirsagar · Answer

yep this makes sense I dont remember how much memory i spawned the instance with so trying to investigate that now I figured I d get it working with local agent first with ample resources and then figure out how to move to docker and then have an automatic build that builds the docker

Rohan Kshirsagar · Answer

regarding the in process config if i already have a parameter to pass in to the op it would look like this ```ops store to table config param blah execution config in process ```

daniel · Answer

exactly

daniel · Answer

that said if the process crashes there too due to a memory issue that may manifest as the run getting stuck in STARTED until you manually terminate it We re working on improving our monitoring capabilities here when the run process crashes the agent will fairly soon detect crashed run worker processes and clean up the state in Dagit

Rohan Kshirsagar · Answer

is this error still indicate of getting into a bad state due to the OOM ```Stack Trace File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud agent dagster cloud agent py line 541 in process api request api result = self handle api request request instance user code launcher File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud agent dagster cloud agent py line 390 in handle api request serialized sensor data or error = client external sensor execution File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster grpc client py line 292 in external sensor execution chunks = list File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster grpc client py line 117 in streaming query raise DagsterUserCodeUnreachableError Could not reach user code server from e```

daniel · Answer

It would be indicative of a process that the agent spins up crashing and no longer being reachable you can tell the agent to spin it back up again by pressing the Redeploy button on the Workspace tab

daniel · Answer

and the most likely reason I can think of for that process crashing would a memory issue on the box

daniel · Answer

the non local agents are better about crash recovery here e g the kubernetes agent runs those servers as k8s deployments so if a process crashes k8s will bring it back up

Rohan Kshirsagar · Answer

i see a python process taking up 1 6G of memory is that a background process that dagster cloud runs

daniel · Answer

Does it say what the command is

Rohan Kshirsagar · Answer

just `python`

daniel · Answer

do you have a way to run `ps aux | grep lt pid gt ` where pid is the pid of that process Usually that gives more information about the arguments that were passed in

Rohan Kshirsagar · Answer

```rohan 7954 4 3 40 5 4303820 1637944 pts 1 Sl+ 16 55 0 13 home rohan miniconda3 envs unify news dag bin python m dagster api grpc lazy load user code socket tmp tmpu4rrl7 x heartbeat heartbeat timeout 60 log level WARNING use python environment entry point d home rohan news bias package name unify news dag```

Rohan Kshirsagar · Answer

i wonder if the loading of all the modules in the package ML NLP heavy causes the memory bloat

Rohan Kshirsagar · Answer

if i start a run i assume that memory will automatically double even before execution

daniel · Answer

Yeah I suspect that s exactly it This actually came up in an issue a couple of days ago <https github com dagster io dagster issues 6824>

daniel · Answer

I d be curious if the workaround suggested there delaying the heavy imports until the ops that actually use them feels like an acceptable workaround

Rohan Kshirsagar · Answer

trade off is that it could hide issues on initialization

daniel · Answer

Yeah exactly

daniel · Answer

If you re running in k8s we have some additional tricks we could pull here for example using different images or python environment for different steps with different imports

Rohan Kshirsagar · Answer

im not sure how dockerizing this will help wouldnt that only add memory

Rohan Kshirsagar · Answer

i would love to run in k8s but as a one man team I m not willing to invest in that yet

daniel · Answer

Dockerizing could help by letting you run each op in its own container and then customizing the image used for each op so that it uses only the imports it needs But that s definitely a more complex setup and in some cases running each op in its own container could result in more memory usage overall So its not a silver bullet

Rohan Kshirsagar · Answer

gotcha where do i find an example of that set up

daniel · Answer

I may have spoken slightly too soon That varying the image per op pattern is currently supported in k8s but not in docker both solutions also wouldn t really help with the memory usage of this standing server I think the first thing I would try would be using the default multiprocess executor plus delaying the beefy imports until the bodies of the ops just to get a baseline number on whether that reduces the memory usage to acceptable levels

Rohan Kshirsagar · Answer

so i got a beefier instance and the job ran successfully I am seeing a transient error ```dagster cloud storage errors GraphQLStorageError Could not find a suitable TLS CA certificate bundle invalid path home rohan miniconda3 envs unify news dag lib python3 8 site packages certifi cacert pem Stack Trace File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud agent dagster cloud agent py line 169 in run loop self check update workspace instance user code launcher File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud agent dagster cloud agent py line 194 in check update workspace self query for workspace updates instance user code launcher upload results=False File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud agent dagster cloud agent py line 242 in query for workspace updates result = instance graphql client execute WORKSPACE ENTRIES QUERY File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud storage client py line 84 in execute raise GraphQLStorageError exc str from exc The above exception was caused by the following exception OSError Could not find a suitable TLS CA certificate bundle invalid path home rohan miniconda3 envs unify news dag lib python3 8 site packages certifi cacert pem Stack Trace File home rohan miniconda3 envs unify news dag lib python3 8 site packages dagster cloud storage client py line 69 in execute response = <http self session post|self session post> self url post args File home rohan miniconda3 envs unify news dag lib python3 8 site packages requests sessions py line 577 in post return self request POST url data=data json=json kwargs File home rohan miniconda3 envs unify news dag lib python3 8 site packages requests sessions py line 529 in request resp = self send prep send kwargs File home rohan miniconda3 envs unify news dag lib python3 8 site packages requests sessions py line 645 in send r = adapter send request kwargs File home rohan miniconda3 envs unify news dag lib python3 8 site packages requests adapters py line 417 in send self cert verify conn request url verify cert File home rohan miniconda3 envs unify news dag lib python3 8 site packages requests adapters py line 228 in cert verify raise IOError Could not find a suitable TLS CA certificate bundle ``` I assume this is a google compute engine setup thing

daniel · Answer

Oh interesting Transient How frequently are you seeing it

Rohan Kshirsagar · Answer

dont see it right now from the cloud cli but i saw it soon after the run happened