Support environment variable substitution #433
Conversation
jbeisen
force-pushed
the
env-var-sub
branch
10 times, most recently
from
20efd7e
to
cab3d45
Compare
jbeisen
force-pushed
the
env-var-sub
branch
3 times, most recently
from
3f9a97a
to
638987c
Compare
| "description": "Comma separated list of headers to send with the request", | ||
| "pattern": "([a-zA-Z0-9-]+: ?.+,)*([a-zA-Z0-9-]+: ?.+)", | ||
| "examples": ["Authentication: digest 1234,Content-Type: application/json"] | ||
| "description": "Comma separated list of headers to send with the request. May contain [variables](https://doc.arroyo.dev/connectors/variables).", |
There was a problem hiding this comment.
I'd expect this link to be something like "https://doc.arroyo.dev/connectors/overview#connection-formats" since we probably wouldn't want an entire doc page just to explain variable substitution
arroyo-types/src/lib.rs
Outdated
| type Value = VarStr; | ||
|
|
||
| fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result { | ||
| formatter.write_str("a string") |
There was a problem hiding this comment.
Yes, this forms the error message if deserialization fails. More details here.
arroyo-types/Cargo.toml
Outdated
| @@ -8,3 +8,5 @@ bincode = "2.0.0-rc.3" | |||
| serde = { version = "1.0", features = ["derive"] } | |||
| arrow = { workspace = true } | |||
| arrow-array = { workspace = true } | |||
| regex = "1.10.2" | |||
There was a problem hiding this comment.
we try to avoid adding dependencies to arroyo-types since it's the base for everything — can this be moved elsewhere? For example, things that are needed by both api and the worker generally live in arroyo-rpc.
| @@ -103,15 +104,15 @@ pub struct ConfluentSchemaRegistry { | |||
| topic: String, | |||
| client: Client, | |||
| api_key: Option<String>, | |||
| api_secret: Option<String>, | |||
| api_secret: Option<VarStr>, | |||
There was a problem hiding this comment.
I'm not sure ConfluentSchemaRegistry should be aware of VarStr — maybe it makes more sense to have the resolution happen earlier?
There was a problem hiding this comment.
My thinking was that we should try to delay variable substitution as long as possible, and do it just in time when it's needed. I think that pattern minimizes the risk that we accidentally log sensitive data.
| @@ -72,12 +72,28 @@ where | |||
| .map(|m| m.to_string()), | |||
| ); | |||
|
|
|||
| let headers = string_to_map( | |||
There was a problem hiding this comment.
Can we abstract this code out into a method so we don't have to repeat it?
In the Kafka, Polling HTTP, WebSocket, and SSE sources, support environment variables in the passwords, secrets, and headers fields. This is done by setting the format of the field in the JSON schema to 'var-str', which tells typify to replace the type with the VarStr struct, which has a function for substituting environment variables. The VarStr struct serializes/deserializes from a JSON string.
In the Kafka, Polling HTTP, WebSocket, and SSE sources, support environment variables in the passwords, secrets, and headers fields.
This is done by setting the
formatof the field in the JSON schema tovar-str, which tells typify to replace the type with theVarStrstruct, which has a function for substituting environment variables. TheVarStrstruct serializes/deserializes from a JSON string.If the variable is needed for testing the connection and it's missing, it'll look like this:
And in SQL it looks like this: