Deployment and lifecycle management for Nix
agent/server: get presigned url and upload seed deployment log
+251
-14
+51
apps/sower/lib/sower/storage.ex
+51
apps/sower/lib/sower/storage.ex
···
1
1
defmodule Sower.Storage do
2
+
require Logger
3
+
4
+
alias SowerClient.Storage.PresignUploadReply
5
+
alias SowerClient.Storage.PresignUploadRequest
6
+
7
+
def presign_upload(%PresignUploadRequest{} = request) do
8
+
method = request.method || "PUT"
9
+
10
+
if method != "PUT" do
11
+
{:error, :unsupported_method}
12
+
else
13
+
opts = presign_upload_opts(request)
14
+
15
+
case presign_upload(request.path, opts) do
16
+
{:ok, url} ->
17
+
{:ok,
18
+
PresignUploadReply.cast!(%{
19
+
url: url,
20
+
method: method,
21
+
headers: presign_upload_headers(request)
22
+
})}
23
+
24
+
{:error, reason} ->
25
+
Logger.error(
26
+
msg: "Failed to presign upload",
27
+
path: request.path,
28
+
method: method,
29
+
reason: inspect(reason)
30
+
)
31
+
32
+
{:error, :failed_to_presign_upload}
33
+
end
34
+
end
35
+
end
36
+
2
37
def presign_upload(file, opts \\ []) do
3
38
bucket = get_in(config(), [:s3, :bucket])
4
39
expires_in = Keyword.get(opts, :expires_in, 60 * 60)
5
40
headers = checksum_headers(opts) ++ Keyword.get(opts, :headers, [])
41
+
42
+
Logger.debug(msg: "Generating presigned url", file: file, expires_in: expires_in)
6
43
7
44
:s3
8
45
|> ExAws.Config.new()
···
19
56
20
57
:error ->
21
58
[]
59
+
end
60
+
end
61
+
62
+
defp presign_upload_opts(%PresignUploadRequest{} = request) do
63
+
case request.checksum_sha256 do
64
+
nil -> []
65
+
checksum -> [checksum_sha256: checksum]
66
+
end
67
+
end
68
+
69
+
defp presign_upload_headers(%PresignUploadRequest{} = request) do
70
+
case request.checksum_sha256 do
71
+
nil -> %{}
72
+
checksum -> %{"x-amz-checksum-sha256" => checksum}
22
73
end
23
74
end
24
75
+2
apps/sower/lib/sower_web/agent_channel.ex
+2
apps/sower/lib/sower_web/agent_channel.ex
···
138
138
Orchestration.update_agent_seed_generations(report, socket.assigns.agent)
139
139
end)
140
140
141
+
handle_schema(SowerClient.Storage.PresignUploadRequest, &Sower.Storage.presign_upload/1)
142
+
141
143
@impl Phoenix.Channel
142
144
def handle_info(:track_presence, %Phoenix.Socket{assigns: %{agent: agent}} = socket) do
143
145
Logger.debug(msg: "Tracking agent presence", agent_sid: agent.sid)
+4
apps/sower_agent/lib/sower_agent/channel_client.ex
+4
apps/sower_agent/lib/sower_agent/channel_client.ex
+38
-1
apps/sower_agent/lib/sower_agent/client.ex
+38
-1
apps/sower_agent/lib/sower_agent/client.ex
···
258
258
{:noreply, socket}
259
259
260
260
deployment ->
261
-
result = SowerAgent.Deployer.run(deployment)
261
+
Task.Supervisor.start_child(SowerAgent.TaskSupervisor, fn ->
262
+
result =
263
+
try do
264
+
SowerAgent.Deployer.run(deployment)
265
+
rescue
266
+
error ->
267
+
Logger.error(
268
+
msg: "Deployment task crashed",
269
+
deployment_sid: deployment.sid,
270
+
error: Exception.format(:error, error, __STACKTRACE__)
271
+
)
272
+
273
+
:failure
274
+
catch
275
+
kind, reason ->
276
+
Logger.error(
277
+
msg: "Deployment task crashed",
278
+
deployment_sid: deployment.sid,
279
+
kind: inspect(kind),
280
+
reason: inspect(reason)
281
+
)
282
+
283
+
:failure
284
+
end
285
+
286
+
send(__MODULE__, {:deployment_completed, deployment.sid, result})
287
+
end)
288
+
289
+
{:noreply, socket}
290
+
end
291
+
end
292
+
293
+
def handle_info({:deployment_completed, sid, result}, socket) do
294
+
case Map.get(socket.active_deployments, sid) do
295
+
nil ->
296
+
Logger.warning(msg: "Deployment not found during completion", sid: sid)
297
+
{:noreply, socket}
262
298
299
+
deployment ->
263
300
{:ok, deployment_result} =
264
301
SowerClient.Orchestration.DeploymentResult.cast(%{
265
302
request_id: deployment.request_id,
+102
-13
apps/sower_agent/lib/sower_agent/deployer.ex
+102
-13
apps/sower_agent/lib/sower_agent/deployer.ex
···
1
1
defmodule SowerAgent.Deployer do
2
2
require Logger
3
3
4
+
alias SowerAgent.Client
4
5
alias SowerAgent.Config
5
6
alias SowerAgent.Storage
6
7
alias SowerClient.Activator
7
8
alias SowerClient.Orchestration.Deployment
8
9
alias SowerClient.Orchestration.DeploymentProfile
9
10
alias SowerClient.Orchestration.SeedDeployment
11
+
alias SowerClient.Storage.PresignUploadReply
12
+
alias SowerClient.Storage.PresignUploadRequest
10
13
11
14
def run(%Deployment{} = deployment) do
12
15
result =
···
305
308
306
309
defp maybe_write_log(_deployment, _seed, []), do: :ok
307
310
308
-
# TODO: when you write to disk, you should ensure it gets deleted
309
311
defp maybe_write_log(%Deployment{} = deployment, seed, output_lines) do
310
-
state_dir = SowerAgent.Config.get().state_directory
311
-
deployments_dir = Path.join(state_dir, "deployments")
312
-
File.mkdir_p!(deployments_dir)
313
-
314
-
date = DateTime.utc_now() |> DateTime.to_unix()
315
-
filename = "#{date}-#{deployment.sid}-#{seed.sid}.log"
316
-
path = Path.join(deployments_dir, filename)
317
-
318
-
cleaned_output =
312
+
content =
319
313
output_lines
320
314
|> Enum.reject(&is_nil/1)
321
315
|> Enum.map(&strip_ansi/1)
316
+
|> Enum.join("\n")
322
317
323
-
content = Enum.join(cleaned_output, "\n")
324
-
File.write!(path, content)
318
+
checksum_sha256 = :crypto.hash(:sha256, content) |> Base.encode64()
319
+
object_path = "logs/deployments/#{deployment.sid}/seeds/#{seed.sid}.log"
325
320
326
-
Logger.debug(msg: "Wrote deployment log", path: path)
321
+
request =
322
+
PresignUploadRequest.cast!(%{
323
+
path: object_path,
324
+
method: "PUT",
325
+
checksum_sha256: checksum_sha256
326
+
})
327
+
328
+
case Client.call(PresignUploadRequest.event(), request, 15_000) do
329
+
{:ok, reply_payload} ->
330
+
case PresignUploadReply.cast(reply_payload) do
331
+
{:ok, reply} ->
332
+
upload_deployment_log(reply, content, deployment.sid, seed.sid, object_path)
333
+
334
+
{:error, error} ->
335
+
Logger.error(
336
+
msg: "Failed to parse presign upload reply",
337
+
deployment_sid: deployment.sid,
338
+
seed_sid: seed.sid,
339
+
object_path: object_path,
340
+
error: inspect(error)
341
+
)
342
+
end
343
+
344
+
{:error, error} ->
345
+
Logger.error(
346
+
msg: "Failed to request presign upload URL",
347
+
deployment_sid: deployment.sid,
348
+
seed_sid: seed.sid,
349
+
object_path: object_path,
350
+
error: inspect(error)
351
+
)
352
+
end
353
+
end
354
+
355
+
defp upload_deployment_log(
356
+
%PresignUploadReply{} = reply,
357
+
content,
358
+
deployment_sid,
359
+
seed_sid,
360
+
object_path
361
+
) do
362
+
case String.upcase(reply.method || "") do
363
+
"PUT" ->
364
+
do_upload_deployment_log(reply, content, deployment_sid, seed_sid, object_path)
365
+
366
+
method ->
367
+
Logger.error(
368
+
msg: "Unsupported presign upload method",
369
+
deployment_sid: deployment_sid,
370
+
seed_sid: seed_sid,
371
+
object_path: object_path,
372
+
method: method
373
+
)
374
+
end
375
+
end
376
+
377
+
defp do_upload_deployment_log(
378
+
%PresignUploadReply{} = reply,
379
+
content,
380
+
deployment_sid,
381
+
seed_sid,
382
+
object_path
383
+
) do
384
+
headers =
385
+
(reply.headers || %{})
386
+
|> Enum.to_list()
387
+
388
+
case Req.put(url: reply.url, headers: headers, body: content, retry: false) do
389
+
{:ok, %Req.Response{status: status}} when status in 200..299 ->
390
+
Logger.info(
391
+
msg: "Uploaded deployment log",
392
+
deployment_sid: deployment_sid,
393
+
seed_sid: seed_sid,
394
+
object_path: object_path
395
+
)
396
+
397
+
{:ok, %Req.Response{status: status, body: body}} ->
398
+
Logger.error(
399
+
msg: "Failed to upload deployment log",
400
+
deployment_sid: deployment_sid,
401
+
seed_sid: seed_sid,
402
+
object_path: object_path,
403
+
status: status,
404
+
body: inspect(body)
405
+
)
406
+
407
+
{:error, error} ->
408
+
Logger.error(
409
+
msg: "Failed to upload deployment log",
410
+
deployment_sid: deployment_sid,
411
+
seed_sid: seed_sid,
412
+
object_path: object_path,
413
+
error: inspect(error)
414
+
)
415
+
end
327
416
end
328
417
329
418
defp strip_ansi(text) do
+2
apps/sower_client/lib/sower_client.ex
+2
apps/sower_client/lib/sower_client.ex
···
21
21
SowerClient.Orchestration.SeedDeployment,
22
22
SowerClient.Orchestration.Subscription,
23
23
SowerClient.Orchestration.SubscriptionSync,
24
+
SowerClient.Storage.PresignUploadReply,
25
+
SowerClient.Storage.PresignUploadRequest,
24
26
SowerClient.Seed,
25
27
SowerClient.SeedMeta,
26
28
SowerClient.SeedTag
+26
apps/sower_client/lib/sower_client/storage/presign_upload_reply.ex
+26
apps/sower_client/lib/sower_client/storage/presign_upload_reply.ex
···
1
+
defmodule SowerClient.Storage.PresignUploadReply do
2
+
use SowerClient.Schema
3
+
4
+
OpenApiSpex.schema(%{
5
+
title: "PresignUploadReply",
6
+
type: :object,
7
+
properties: %{
8
+
url: %Schema{
9
+
type: :string,
10
+
description: "Signed URL for direct upload"
11
+
},
12
+
method: %Schema{
13
+
type: :string,
14
+
description: "HTTP method to use for upload",
15
+
default: "PUT"
16
+
},
17
+
headers: %Schema{
18
+
type: :object,
19
+
description: "Headers that must be sent with upload request",
20
+
default: %{},
21
+
additionalProperties: %Schema{type: :string}
22
+
}
23
+
},
24
+
required: [:url, :method, :headers]
25
+
})
26
+
end
+26
apps/sower_client/lib/sower_client/storage/presign_upload_request.ex
+26
apps/sower_client/lib/sower_client/storage/presign_upload_request.ex
···
1
+
defmodule SowerClient.Storage.PresignUploadRequest do
2
+
use SowerClient.Schema
3
+
use SowerClient.ChannelMessage, event: "storage:presign_upload"
4
+
5
+
OpenApiSpex.schema(%{
6
+
title: "PresignUploadRequest",
7
+
type: :object,
8
+
properties: %{
9
+
path: %Schema{
10
+
type: :string,
11
+
description: "Path of object in storage bucket"
12
+
},
13
+
method: %Schema{
14
+
type: :string,
15
+
description: "HTTP method to be used with signed URL",
16
+
default: "PUT"
17
+
},
18
+
checksum_sha256: %Schema{
19
+
type: :string,
20
+
description: "Base64 SHA-256 checksum to sign for upload integrity",
21
+
nullable: true
22
+
}
23
+
},
24
+
required: [:path]
25
+
})
26
+
end