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,12 @@ 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 = {}
+        if os.getenv("ANYSCALE__AIRFLOW_TELEMETRY_ENABLED", "true") == "false":
+            headers["X-Anyscale-Source"] = "astronomer-sdk"
+
+        self.sdk = Anyscale(auth_token=token, headers=headers)
 
     @classmethod
     def get_ui_field_behaviour(cls) -> Dict[str, Any]: