Feature/ingest v2

data/pom.xml

@@ -258,4 +258,4 @@
             <scope>test</scope>
         </dependency>
     </dependencies>
-</project>
+</project>

ingest-v2/README.md

@@ -0,0 +1,38 @@
+# ingest-v2
+
+This project was created using the [Ktor Project Generator](https://start.ktor.io).
+
+Here are some useful links to get you started:
+
+- [Ktor Documentation](https://ktor.io/docs/home.html)
+- [Ktor GitHub page](https://github.com/ktorio/ktor)
+- The [Ktor Slack chat](https://app.slack.com/client/T09229ZC6/C0A974TJ9). You'll need to [request an invite](https://surveys.jetbrains.com/s3/kotlin-slack-sign-up) to join.
+
+## Features
+
+Here's a list of features included in this project:
+
+| Name                                                                   | Description                                                                        |
+|------------------------------------------------------------------------|------------------------------------------------------------------------------------|
+| [Content Negotiation](https://start.ktor.io/p/content-negotiation)     | Provides automatic content conversion according to Content-Type and Accept headers |
+| [Routing](https://start.ktor.io/p/routing)                             | Provides a structured routing DSL                                                  |
+| [kotlinx.serialization](https://start.ktor.io/p/kotlinx-serialization) | Handles JSON serialization using kotlinx.serialization library                     |
+| [AsyncAPI](https://start.ktor.io/p/asyncapi)                           | Generates and serves AsyncAPI documentation                                        |
+
+## Building & Running
+
+To build or run the project, use one of the following tasks:
+
+| Task                                                         | Description       |
+|--------------------------------------------------------------|-------------------|
+| `mvn test`                                                   | Run the tests     |
+| `mvn package`                                                | Build the project |
+| `java -jar target/ingest-v2-0.0.1-jar-with-dependencies.jar` | Run the server    |
+
+If the server starts successfully, you'll see the following output:
+
+```
+2024-12-04 14:32:45.584 [main] INFO  Application - Application started in 0.303 seconds.
+2024-12-04 14:32:45.682 [main] INFO  Application - Responding at http://0.0.0.0:8080
+```
+

ingest-v2/src/main/kotlin/com/microsoft/azure/kusto/ingest/v2/ConfigurationClient.kt

@@ -0,0 +1,105 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+package com.microsoft.azure.kusto.ingest.v2
+
+import com.azure.core.credential.TokenCredential
+import com.microsoft.azure.kusto.ingest.v2.common.exceptions.IngestException
+import com.microsoft.azure.kusto.ingest.v2.common.models.ClientDetails
+import com.microsoft.azure.kusto.ingest.v2.common.models.S2SToken
+import com.microsoft.azure.kusto.ingest.v2.infrastructure.HttpResponse
+import com.microsoft.azure.kusto.ingest.v2.models.ConfigurationResponse
+import io.ktor.http.HttpStatusCode
+import org.slf4j.LoggerFactory
+import java.net.ConnectException
+
+class ConfigurationClient(
+    override val dmUrl: String,
+    override val tokenCredential: TokenCredential,
+    override val skipSecurityChecks: Boolean = false,
+    override val clientDetails: ClientDetails,
+    override val s2sTokenProvider: (suspend () -> S2SToken)? = null,
+    override val s2sFabricPrivateLinkAccessContext: String? = null,
+) :
+    KustoBaseApiClient(
+        dmUrl,
+        tokenCredential,
+        skipSecurityChecks,
+        clientDetails,
+        s2sTokenProvider = s2sTokenProvider,
+        s2sFabricPrivateLinkAccessContext = s2sFabricPrivateLinkAccessContext,
+    ) {
+    private val logger =
+        LoggerFactory.getLogger(ConfigurationClient::class.java)
+    private val baseUrl = "$dmUrl/v1/rest/ingestion/configuration"
+
+    suspend fun getConfigurationDetails(): ConfigurationResponse {
+        try {
+            val configurationHttpResponse: HttpResponse<ConfigurationResponse> =
+                api.getIngestConfiguration()
+            if (configurationHttpResponse.success) {
+                logger.info(
+                    "Successfully retrieved configuration details from $dmUrl with status: ${configurationHttpResponse.status}",
+                )
+                logger.debug(
+                    "Configuration details: {}",
+                    configurationHttpResponse.body(),
+                )
+                return configurationHttpResponse.body()
+            } else if (
+                configurationHttpResponse.status ==
+                HttpStatusCode.NotFound.value
+            ) {
+                /*
+                404 is a special case - it indicates that the endpoint is not found. This may be a transient
+                network issue
+                 */
+                val message =
+                    "Endpoint $dmUrl not found. Please ensure the cluster supports queued ingestion."
+                logger.error(
+                    "{}. Status: {}",
+                    message,
+                    configurationHttpResponse.status,
+                )
+                throw IngestException(
+                    message = message,
+                    cause = ConnectException(message),
+                    failureCode = configurationHttpResponse.status,
+                    failureSubCode = "",
+                    isPermanent = false,
+                )
+            } else {
+                val configurationResponseBody = configurationHttpResponse.body()
+                val message =
+                    "Failed to retrieve configuration details from $baseUrl.Status: ${configurationHttpResponse.status}, " +
+                        "Body: $configurationResponseBody"
+                logger.error("{}", message)
+                throw IngestException(
+                    message = message,
+                    failureCode = configurationHttpResponse.status,
+                )
+            }
+        } catch (notAbleToReachHost: ConnectException) {
+            val message =
+                "Failed to reach $baseUrl. Please ensure the cluster address is correct and the cluster is reachable."
+            throw IngestException(
+                message = message,
+                cause = notAbleToReachHost,
+                failureCode = HttpStatusCode.NotFound.value,
+                failureSubCode = "",
+                isPermanent = false,
+            )
+        } catch (ex: Exception) {
+            if (ex is IngestException) throw ex
+            val message =
+                "An unexpected error occurred while trying to reach $baseUrl"
+            throw IngestException(
+                message = message,
+                cause = ex,
+                // Mark this as a 5xx series error
+                failureCode = HttpStatusCode.InternalServerError.value,
+                failureSubCode = "",
+                isPermanent = true,
+            )
+        }
+    }
+}

ingest-v2/src/main/kotlin/com/microsoft/azure/kusto/ingest/v2/IngestV2.kt

@@ -0,0 +1,77 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+package com.microsoft.azure.kusto.ingest.v2
+
+// Size of each block to upload to Azure Blob Storage (4 MB)
+const val UPLOAD_BLOCK_SIZE_BYTES: Long = 4 * 1024 * 1024
+
+// Maximum size for a single upload operation to Azure Blob Storage (256 MB)
+const val UPLOAD_MAX_SINGLE_SIZE_BYTES: Long = 256 * 1024 * 1024
+
+// Request timeout in milliseconds for Kusto API HTTP requests
+const val KUSTO_API_REQUEST_TIMEOUT_MS: Long = 60_000
+
+// Connection timeout in milliseconds for Kusto API HTTP requests
+const val KUSTO_API_CONNECT_TIMEOUT_MS: Long = 60_000
+
+// Socket timeout in milliseconds for Kusto API HTTP requests
+const val KUSTO_API_SOCKET_TIMEOUT_MS: Long = 60_000
+
+// Kusto API version used in HTTP requests
+const val KUSTO_API_VERSION = "2024-12-12"
+
+// Default refresh interval for configuration cache (1 hour)
+const val CONFIG_CACHE_DEFAULT_REFRESH_INTERVAL_HOURS: Long = 1
+
+// Default value for skipSecurityChecks if not provided
+const val CONFIG_CACHE_DEFAULT_SKIP_SECURITY_CHECKS: Boolean = false
+
+// Default interval between retries for SimpleRetryPolicy (10 seconds)
+const val INGEST_RETRY_POLICY_DEFAULT_INTERVAL_SECONDS: Long = 10
+
+// Default total number of retries for SimpleRetryPolicy
+const val INGEST_RETRY_POLICY_DEFAULT_TOTAL_RETRIES: Int = 3
+
+// Default timeout for blob upload operations (1 hour)
+const val BLOB_UPLOAD_TIMEOUT_HOURS: Long = 1
+
+// Default retry intervals for CustomRetryPolicy (1s, 3s, 7s)
+val INGEST_RETRY_POLICY_CUSTOM_INTERVALS: Array<Long> = arrayOf(1, 3, 7)
+
+// Number of blobs to upload in a single batch
+const val MAX_BLOBS_PER_BATCH: Int = 70
+
+// Default maximum data size for blob upload operations (4GB)
+const val UPLOAD_CONTAINER_MAX_DATA_SIZE_BYTES: Long = 4L * 1024 * 1024 * 1024
+
+// Default maximum concurrency for blob upload operations
+const val UPLOAD_CONTAINER_MAX_CONCURRENCY: Int = 4
+
+const val STREAMING_MAX_REQ_BODY_SIZE = 10 * 1024 * 1024 // 10 MB
+
+// Managed Streaming Policy Defaults
+
+// Default value for continueWhenStreamingIngestionUnavailable in ManagedStreamingPolicy
+// When false, the client will fail if streaming ingestion is unavailable
+const val MANAGED_STREAMING_CONTINUE_WHEN_UNAVAILABLE_DEFAULT: Boolean = false
+
+// Default data size factor for ManagedStreamingPolicy
+// Factor used to determine size threshold for queued ingestion (1.0 = no adjustment)
+const val MANAGED_STREAMING_DATA_SIZE_FACTOR_DEFAULT: Double = 1.0
+
+// Default throttle backoff period in seconds for ManagedStreamingPolicy
+// How long to use queued ingestion after streaming is throttled
+const val MANAGED_STREAMING_THROTTLE_BACKOFF_SECONDS: Long = 10
+
+// Default time until resuming streaming ingestion in minutes for ManagedStreamingPolicy
+// How long to use queued ingestion after streaming becomes unavailable
+const val MANAGED_STREAMING_RESUME_TIME_MINUTES: Long = 15
+
+// Default retry delays for ManagedStreamingPolicy (in seconds): 1s, 2s, 4s
+val MANAGED_STREAMING_RETRY_DELAYS_SECONDS: Array<Long> = arrayOf(1, 2, 4)
+
+// Maximum jitter to add to retry delays in milliseconds
+const val MANAGED_STREAMING_RETRY_JITTER_MS: Long = 1000
+
+const val STREAM_COMPRESSION_BUFFER_SIZE_BYTES: Int = 64 * 1024
+const val STREAM_PIPE_BUFFER_SIZE_BYTES: Int = 1024 * 1024