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

@@ -46,7 +46,14 @@ def client(self) -> Anyscale:
         if not token:
             raise AirflowException(f"Missing API token for connection id {self.conn_id}")
 
-        return 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"] = "airflow"
+
+        return Anyscale(auth_token=token, headers=headers)
 
     @classmethod
     def get_ui_field_behaviour(cls) -> dict[str, Any]: