Add telemetry headers

- Capture telemetry via a header at SDK init
- Enable users to opt-out of telemetry via setting env_var

anyscale_provider/hooks/anyscale.py

@@ -120,8 +120,14 @@ def __init__(self, conn_id: str = default_conn_name, **kwargs: Any) -> None:
         if not token:
             raise AirflowException(f"Missing API token for connection id {self.conn_id}")
 
-        self.sdk = Anyscale(auth_token=token)
-
+        # Add custom headers if telemetry is enabled - by default telemetry is enabled.
+        headers = {}
+        telemetry_env = os.getenv("ANYSCALE__AIRFLOW_TELEMETRY_ENABLED", "true")
+        telemetry_enabled = telemetry_env.lower() in ["true", "1", "yes", "on"]
+        if telemetry_enabled:
+            headers["X-Anyscale-Source"] = "astronomer-sdk"
+
+        self.sdk = Anyscale(auth_token=token, headers=headers)
 
     @classmethod
     def get_ui_field_behaviour(cls) -> Dict[str, Any]: