diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Bundle.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Bundle.kt index 6cd549270a8a4..5445db5c5ee93 100644 --- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Bundle.kt +++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Bundle.kt @@ -25,6 +25,11 @@ package org.apache.airflow.sdk * Build a [Bundle] by implementing [BundleBuilder], then pass it to * [Server.serve] to start accepting task-execution requests. * + * Dag and task IDs the Airflow server would reject (longer than 250 + * characters, or containing anything other than letters, digits, dashes, + * dots, and underscores) produce a best-effort warning in the task logs; + * the server remains the authoritative validator. + * * @property dags All registered Dags keyed by [DagDef.id]. * @throws IllegalArgumentException if any two Dags share the same ID. */ @@ -32,6 +37,10 @@ class Bundle( dags: Iterable, ) { internal val dags: Map = dags.associateByDagId() + + init { + IdValidation.warnOnSuspiciousIds(this.dags.values) + } } private fun Iterable.associateByDagId(): Map { diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt index e3ba700a0fa1c..d93182b76ccd8 100644 --- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt +++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt @@ -36,13 +36,16 @@ import kotlin.Throws * .addTask("load", Load.class); * ``` * - * @param id Dag identifier. Must contain only ASCII alphanumeric characters, - * dashes, dots, or underscores; must be unique within a [Bundle]. + * Dag and task IDs are validated authoritatively by the Airflow server. This + * class accepts any ID; IDs the server would reject produce a best-effort + * warning when the Dags are assembled into a [Bundle]. + * + * @param id Dag identifier; must be unique within a [Bundle]. * * @see Builder.Dag */ class DagDef( - val id: String, // TODO: charset check? + val id: String, ) { internal val tasks = linkedMapOf() diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/IdValidation.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/IdValidation.kt new file mode 100644 index 0000000000000..d9f573c631781 --- /dev/null +++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/IdValidation.kt @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.airflow.sdk + +import org.apache.airflow.sdk.execution.Logger + +/** + * Best-effort checks of Dag and task IDs against the rules the Airflow server + * enforces (`airflow.utils.helpers.validate_key`). Warn-only: the server + * validates authoritatively, and checks like the `..` one depend on server + * configuration this process cannot see. + */ +internal object IdValidation { + private const val MAX_ID_LENGTH = 250 + private val ID_REGEX = Regex("""^[\p{L}\p{N}_.-]+$""") + + private val logger = Logger(IdValidation::class) + + fun warnOnSuspiciousIds(dags: Iterable) { + for (dag in dags) { + warnOnSuspiciousId("Dag id", dag.id, mapOf("dag_id" to dag.id)) + for (taskId in dag.tasks.keys) { + warnOnSuspiciousId("Task id", taskId, mapOf("dag_id" to dag.id, "task_id" to taskId)) + } + } + } + + private fun warnOnSuspiciousId( + label: String, + id: String, + arguments: Map, + ) { + val length = id.codePointCount(0, id.length) + if (length > MAX_ID_LENGTH) { + logger.warning( + "$label is longer than $MAX_ID_LENGTH characters; the Airflow server will reject it", + arguments + ("length" to length), + ) + } + if (!ID_REGEX.matches(id)) { + logger.warning( + "$label must be made of alphanumeric characters, dashes, dots, and underscores; " + + "the Airflow server will reject it", + arguments, + ) + } else if (id.contains("..")) { + logger.warning( + "$label contains '..'; the Airflow server will reject it " + + "unless [core] allow_double_dot_in_ids is enabled", + arguments, + ) + } + } +} diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/BundleTest.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/BundleTest.kt index 57050754e887e..5d23e95c07631 100644 --- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/BundleTest.kt +++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/BundleTest.kt @@ -19,11 +19,26 @@ package org.apache.airflow.sdk +import org.apache.airflow.sdk.execution.Level +import org.apache.airflow.sdk.execution.LogCapture import org.junit.jupiter.api.Assertions +import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.DisplayName import org.junit.jupiter.api.Test internal class BundleTest { + private class NoOp : Task { + override fun execute( + context: Context, + client: Client, + ) = Unit + } + + @BeforeEach + fun setUp() { + LogCapture.drain() + } + @Test @DisplayName("Should index dags by dagId") fun shouldIndexDagsByDagId() { @@ -32,6 +47,22 @@ internal class BundleTest { val bundle = Bundle(listOf(dag)) Assertions.assertEquals(mapOf("dag" to dag), bundle.dags) + Assertions.assertEquals(emptyList(), LogCapture.drain()) + } + + @Test + @DisplayName("Should warn on dag and task ids the server would reject") + fun shouldWarnOnSuspiciousIds() { + Bundle(listOf(DagDef("bad dag").addTask("bad task", NoOp::class.java))) + + val warnings = LogCapture.drain() + Assertions.assertEquals(2, warnings.size) + Assertions.assertEquals(setOf(Level.WARNING), warnings.map { it.level }.toSet()) + Assertions.assertEquals(mapOf("dag_id" to "bad dag"), warnings[0].arguments) + Assertions.assertEquals( + mapOf("dag_id" to "bad dag", "task_id" to "bad task"), + warnings[1].arguments, + ) } @Test diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/IdValidationTest.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/IdValidationTest.kt new file mode 100644 index 0000000000000..845562579af70 --- /dev/null +++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/IdValidationTest.kt @@ -0,0 +1,127 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.airflow.sdk + +import org.apache.airflow.sdk.execution.CapturedLogMessage +import org.apache.airflow.sdk.execution.Level +import org.apache.airflow.sdk.execution.LogCapture +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.DisplayName +import org.junit.jupiter.api.Test + +private const val LOGGER_NAME = "org.apache.airflow.sdk.IdValidation" + +private fun dagTooLongWarning( + id: String, + length: Int, +) = CapturedLogMessage( + Level.WARNING, + LOGGER_NAME, + "Dag id is longer than 250 characters; the Airflow server will reject it", + mapOf("dag_id" to id, "length" to length), +) + +private fun dagCharsetWarning(id: String) = + CapturedLogMessage( + Level.WARNING, + LOGGER_NAME, + "Dag id must be made of alphanumeric characters, dashes, dots, and underscores; " + + "the Airflow server will reject it", + mapOf("dag_id" to id), + ) + +private fun dagDoubleDotWarning(id: String) = + CapturedLogMessage( + Level.WARNING, + LOGGER_NAME, + "Dag id contains '..'; the Airflow server will reject it " + + "unless [core] allow_double_dot_in_ids is enabled", + mapOf("dag_id" to id), + ) + +internal class IdValidationTest { + private class NoOp : Task { + override fun execute( + context: Context, + client: Client, + ) = Unit + } + + @BeforeEach + fun setUp() { + LogCapture.drain() + } + + @Test + @DisplayName("dag id warnings — exact events across every branch") + fun dagIdWarnings() { + val astral = "𠀀" + val tooLongAndInvalid = "a".repeat(250) + " b" + val cases: List>> = + listOf( + "simple" to emptyList(), + "with-dash" to emptyList(), + "with.dot" to emptyList(), + "with_underscore" to emptyList(), + "0numeric" to emptyList(), + "café_dag" to emptyList(), + "任務" to emptyList(), + "a".repeat(250) to emptyList(), + astral.repeat(250) to emptyList(), + "a".repeat(251) to listOf(dagTooLongWarning("a".repeat(251), 251)), + "任".repeat(251) to listOf(dagTooLongWarning("任".repeat(251), 251)), + astral.repeat(251) to listOf(dagTooLongWarning(astral.repeat(251), 251)), + "with space" to listOf(dagCharsetWarning("with space")), + "with/slash" to listOf(dagCharsetWarning("with/slash")), + "with:colon" to listOf(dagCharsetWarning("with:colon")), + "with\ttab" to listOf(dagCharsetWarning("with\ttab")), + "a..b c" to listOf(dagCharsetWarning("a..b c")), + "a..b" to listOf(dagDoubleDotWarning("a..b")), + tooLongAndInvalid to + listOf(dagTooLongWarning(tooLongAndInvalid, 252), dagCharsetWarning(tooLongAndInvalid)), + ) + cases.forEach { (id, expected) -> + IdValidation.warnOnSuspiciousIds(listOf(DagDef(id))) + assertEquals(expected, LogCapture.drain(), "id=$id") + } + } + + @Test + @DisplayName("a task warning carries its dag id") + fun taskWarningCarriesDagId() { + val dag = DagDef("my_dag").addTask("bad task", NoOp::class.java) + + IdValidation.warnOnSuspiciousIds(listOf(dag)) + + assertEquals( + listOf( + CapturedLogMessage( + Level.WARNING, + LOGGER_NAME, + "Task id must be made of alphanumeric characters, dashes, dots, and underscores; " + + "the Airflow server will reject it", + mapOf("dag_id" to "my_dag", "task_id" to "bad task"), + ), + ), + LogCapture.drain(), + ) + } +}