o ưi`=@sddlZddlZddlZddlZddlZddlmZddlmZmZm Z ddl m Z m Z m Z mZmZmZddlmZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lTdd l m!Z!e rqddl"m#Z#ne Z#GdddeeZ$dS)N)uuid)datetime timedeltatimezone) TYPE_CHECKINGAnyDictListOptionalTuple)quote)verbose_logger)LITELLM_ASYNCIO_QUEUE_MAXSIZE)AdditionalLoggingUtils) GCSBucketBase)CommonProxyErrors)IntegrationHealthCheckStatus)*)StandardLoggingPayload) VertexBasec seZdZd) r r?r@r0r-rArBrCrDrErFr')r3r<r=rGrHrIrJr6r6r7async_log_failure_eventWs. "z'GCSBucketLogger.async_log_failure_eventcCsPg}t||jkr&z ||jWn tjyY|Swt||jks |S)a Drain items from the queue (non-blocking), respecting batch_size limit. This prevents unbounded queue growth when processing is slower than log accumulation. Returns: List of items to process, up to batch_size items )lenrappendr- get_nowaitr* QueueEmpty)r3items_to_processr6r6r7_drain_queue_batchps z"GCSBucketLogger._drain_queue_batchdate_strbatch_idcCs|d|dS)zm Generate object name for a batched log file. Format: {date}/batch-{batch_id}.ndjson z/batch-z.ndjsonr6)r3rSrTr6r6r7_generate_batch_object_namesz+GCSBucketLogger._generate_batch_object_namer<cCsJ|ddpi}|ddp|jpd}|ddp|jpd}|d|S)a Extract a synchronous grouping key from kwargs to group items by GCS config. This allows us to batch items with the same bucket/credentials together. Returns a string key that uniquely identifies the GCS config combination. This key may contain sensitive information (bucket names, paths) - use _sanitize_config_key() for logging purposes. standard_callback_dynamic_paramsNZgcs_bucket_namedefaultZgcs_path_service_account|)r@Z BUCKET_NAMEZpath_service_account_json)r3r<rVrpath_service_accountr6r6r7_get_config_keys zGCSBucketLogger._get_config_key config_keycCs&t|d}d|ddS)z Create a sanitized version of the config key for logging. Uses a hash to avoid exposing sensitive bucket names or service account paths. Returns a short hash prefix for safe logging. zutf-8zconfig-N)hashlibsha256encode hexdigest)r3r[Zhash_objr6r6r7_sanitize_config_keysz$GCSBucketLogger._sanitize_config_keyitemscCs>i}|D]}||d}||vrg||<|||q|S)z Group items by their GCS config (bucket + credentials). This ensures items with different configs are processed separately. Returns a dict mapping config_key -> list of items with that config. r<)rZrN)r3rbgroupeditemr[r6r6r7_group_items_by_configsz&GCSBucketLogger._group_items_by_configcCs:g}|D]}|d}tj|tdd}||qd|S)z Combine multiple log payloads into newline-delimited JSON (NDJSON) format. Each line is a valid JSON object representing one log entry. r;F)rW ensure_ascii )jsondumpsr'rNjoin)r3rblinesrdrIZ json_liner6r6r7_combine_payloads_to_ndjsons   z+GCSBucketLogger._combine_payloads_to_ndjsonc s|sdS|dd}zY||IdH}|j|d|ddIdH}|d}|ttj}ttd d t j dd }| ||} | |} |j||| | d IdHt|} d} | | fWSty} zd} t|} td t| | | fWYd} ~ Sd} ~ ww)z Send a batch of items that share the same GCS config. Returns: (success_count, error_count) )rrrr<Nvertex_instancerYrmZservice_account_jsonri-r\headersr object_namerIz6GCS Bucket error logging batch payload to GCS bucket: )get_gcs_logging_configconstruct_request_headers_get_object_date_from_datetimernowrutcr$timeruuid4hexrUrl_log_json_data_on_gcsrMrEr rFr')r3rbr[Z first_kwargsgcs_logging_configrqr current_daterTrrZcombined_payloadZ success_countZ error_countrJr6r6r7_send_grouped_batchsF   (     z#GCSBucketLogger._send_grouped_batchcs |D] }||IdHqdS)z Send each log individually as separate GCS objects (legacy behavior). This is used when GCS_USE_BATCHED_LOGGING is disabled. N)_send_single_log_item)r3rbrdr6r6r7_send_individual_logssz%GCSBucketLogger._send_individual_logsrdc sz;||dIdH}|j|d|ddIdH}|d}|j|d|d|dd }|j||||dd IdHWdStyY}ztd t|WYd}~dSd}~ww) zH Send a single log item to GCS as an individual object. r<NrmrYrnrr;r=)r<rIr=rpz;GCS Bucket error logging individual payload to GCS bucket: )rsrt_get_object_namer{rEr rFr')r3rdr|rqrrrrJr6r6r7rs6   z%GCSBucketLogger._send_single_log_itemcs^|}|s dS|jr%||}|D] \}}|||IdHqdS||IdHdS)aQ Process queued logs - sends logs to GCS Bucket. If `GCS_USE_BATCHED_LOGGING` is enabled (default), batches multiple log payloads into single GCS object uploads (NDJSON format), dramatically reducing API calls. If disabled, sends each log individually as separate GCS objects (legacy behavior). N)rRr)rerbr~r)r3rQZ grouped_itemsr[Z group_itemsr6r6r7async_send_batchs  z GCSBucketLogger.async_send_batchrIr=cCsz|ttj}|dddur|j|d}n |j||ddd}|ddp*i}|ddp2i}d |vr;|d }|S) zD Get the object name to use for the current payload Z error_strN)request_date_stridr response_idZlitellm_paramsmetadataZ gcs_log_id)rurrvrrwr@_generate_failure_object_name_generate_success_object_name)r3r<rIr=r}rrZ_litellm_params _metadatar6r6r7r*s z GCSBucketLogger._get_object_name request_idstart_time_utc end_time_utcc s|dur td||tdd|tddg}d}|D]L}z+|j|d}|j||d}t|dd}||IdH} | durHt| } | WSWqtyh} zt d |d t | WYd} ~ qd} ~ wwdS) z Get the request and response payload for a given `request_id` Tries current day, next day, and previous day until it finds the payload Nz@start_time_utc is required for getting a payload from GCS Bucket)days) datetime_objrr)safez!Failed to fetch payload for date z: ) r0rrurr Zdownload_gcs_objectrhloadsrEr r?r') r3rrrZ dates_to_tryrSdaterrZencoded_object_nameresponseZloaded_responserJr6r6r7get_request_response_payloadCs>        z,GCSBucketLogger.get_request_response_payloadrrcCs|d|S)N/r6)r3rrr6r6r7rmsz-GCSBucketLogger._generate_success_object_namecCs|dtjS)Nz /failure-)rryrz)r3rr6r6r7rtsz-GCSBucketLogger._generate_failure_object_namercCs |dS)Nz%Y-%m-%d)strftime)r3rr6r6r7ruzs z.GCSBucketLogger._get_object_date_from_datetimecs|IdHt|_dS)zB Override flush_queue to work with asyncio.Queue. N)rrxZlast_flush_timer3r6r6r7rB}szGCSBucketLogger.flush_queuecs: t|jIdHtd|jd|IdHq)zE Override periodic_flush to work with asyncio.Queue. TNz GCS Bucket periodic flush after z seconds)r*sleeprr r?rBrr6r6r7r/s zGCSBucketLogger.periodic_flushcs td)Nz(GCS Bucket does not support health check)NotImplementedErrorrr6r6r7async_health_checksz"GCSBucketLogger.async_health_check)N)%__name__ __module__ __qualname__r r'r#rKrLr rDrRrUrrrZrarerlr r$r~rrrrrrdictrrrrurBr/rr __classcell__r6r6r4r7rs`" " /   *   r)%r*r]rhr%rxZ litellm._uuidrrrrtypingrrrr r r urllib.parser Zlitellm._loggingr Zlitellm.constantsrZ-litellm.integrations.additional_logging_utilsrZ/litellm.integrations.gcs_bucket.gcs_bucket_baserZlitellm.proxy._typesrZ,litellm.types.integrations.base_health_checkrZ%litellm.types.integrations.gcs_bucketZlitellm.types.utilsrZ&litellm.llms.vertex_ai.vertex_llm_baserrr6r6r6r7s*