tPj<ddlZddlZddlZddlZddlZddlmZmZmZddlm Z m Z m Z ddl m Z ddlmZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd ldd lmZe rddlm Z ne Z GddeeZ!y)N)datetime timedeltatimezone) TYPE_CHECKINGAnyFinal)verbose_logger)uuid)LITELLM_ASYNCIO_QUEUE_MAXSIZE)AdditionalLoggingUtils) GCSBucketBase)sanitize_cloud_object_component)CommonProxyErrors)IntegrationHealthCheckStatus)*)StandardLoggingPayload) VertexBasec eZdZd'dedzddffd ZdZdZdeefdZ ded edefd Z d e ee fdefd Z d edefdZdeede eeeffdZdeedefdZdeed edeeeffdZdeeddfdZdeddfdZdZd e dede defdZdededzdedzde dzfdZdededefd Zdedefd!Zd"edefd#Zd$Zd%Zde fd&Z!xZ"S)(GCSBucketLoggerN bucket_namereturncddlm}t| |t t j dt|_t t j dt|_ t j dttjjdk(|_tj |_t| |j"|j|jtj$t& |_tj*|j-t/j||d ur&t1d t2j4j6y) Nr premium_user)rGCS_BATCH_SIZEGCS_FLUSH_INTERVALGCS_USE_BATCHED_LOGGINGtrue) flush_lock batch_sizeflush_interval)maxsizeTCGCS Bucket logging is a premium feature. Please upgrade to use it. )litellm.proxy.proxy_serverrsuper__init__intosgetenvGCS_DEFAULT_BATCH_SIZEr "GCS_DEFAULT_FLUSH_INTERVAL_SECONDSr!strGCS_DEFAULT_USE_BATCHED_LOGGINGloweruse_batched_loggingasyncioLockrQueuer log_queue create_taskperiodic_flushr ValueErrorrnot_premium_uservalue)selfrr __class__s o/home/oem/CyberBrain26-unified/.venv/lib/python3.12/site-packages/litellm/integrations/gcs_bucket/gcs_bucket.pyr&zGCSBucketLogger.__init__s$; [1bii(8:PQR!")),@Bd"ef II/5T1U1[1[1] ^ d d fjp p  ",,. ..  :AOl9mD//12''- t #UVgVxVxV~V~UA  $cKddlm}|dur&tdtjj  t jd|||jdd}| td|jjr|jd{|jjt|||d{y747#t$r }t jd |Yd}~yd}~wwxYww) NrrTr#zHGCS Logger: async_log_success_event logging kwargs: %s, response_obj: %sstandard_logging_object+standard_logging_object not found in kwargspayloadkwargs response_objGCS Bucket logging error: %s)r$rr6rr7r8r debuggetr3full flush_queueputGCSLogQueueItem Exception exception)r9rBrC start_timeend_timerlogging_payloades r;async_log_success_eventz'GCSBucketLogger.async_log_success_event7s; t #UVgVxVxV~V~UA  H  Z  EKJJOhjnDoO& !NOO~~""$&&(((..$$__U[jv%wx x x) x H  $ $%CQ G G HsS1C<A#CC .CCC C< CC C9C4/C<4C99C<cK tjd|||jdd}| td|jj r|j d{|jjt|||d{y747#t$r }tjd|Yd}~yd}~wwxYww)NzHGCS Logger: async_log_failure_event logging kwargs: %s, response_obj: %sr>r?r@rD) r rErFr6r3rGrHrIrJrKrL)r9rBrCrMrNrOrPs r;async_log_failure_eventz'GCSBucketLogger.async_log_failure_eventOs H  Z  EKJJOhjnDoO& !NOO~~""$&&(((..$$__U[jv%wx x x) x H  $ $%CQ G G HsSC A#B 'B(.B BB C B B C )C?C C  C cg}t||jkrC |j|jj t||jkrC|S#t j $rY|SwxYw)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 )lenr appendr3 get_nowaitr0 QueueEmpty)r9items_to_processs r;_drain_queue_batchz"GCSBucketLogger._drain_queue_batchbsx:<"#doo5  ''(A(A(CD"#doo5  %%  s)A A76A7date_strbatch_idc|d|dS)zm Generate object name for a batched log file. Format: {date}/batch-{batch_id}.ndjson z/batch-z.ndjson)r9r[r\s r;_generate_batch_object_namez+GCSBucketLogger._generate_batch_object_namess 78*G44r<rBc|jddxsi}|jddxs|jxsd}|jddxs|jxsd}|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_paramsNgcs_bucket_namedefaultgcs_path_service_account|)rF BUCKET_NAMEpath_service_account_json)r9rBrarpath_service_accounts r;_get_config_keyzGCSBucketLogger._get_config_keyzs39**=_ae2f2ljl(6::;LdStW[WgWgtkt , 0 01KT R --   a 4566r< config_keycvtj|jd}d|jddS)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)r9rjhash_objs r;_sanitize_config_keyz$GCSBucketLogger._sanitize_config_keys;"..):):7)CD++-bq1233r<itemsczi}|D]3}|j|d}||vrg||<||j|5|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. rB)rirV)r9rsgroupeditemrjs r;_group_items_by_configz&GCSBucketLogger._group_items_by_configsX<> -D--d8n=J(&( # J  & &t ,  - r<cg}|D]4}|d}tj|td}|j|6dj |S)z Combine multiple log payloads into newline-delimited JSON (NDJSON) format. Each line is a valid JSON object representing one log entry. rAF)rc ensure_ascii )jsondumpsr,rVjoin)r9rslinesrvrO json_lines r;_combine_payloads_to_ndjsonz+GCSBucketLogger._combine_payloads_to_ndjsonsP  $D"9oO ?CeTI LL # $yyr<cK|sy|dd} |j|d{}|j|d|dd{}|d}|jtjt j }ttjd zd tjjdd }|j||} |j|} |j||| | d{t|} d} | | fS777#t $r1} d} t|} t#j$d | | | fcYd} ~ Sd} ~ wwxYww)z Send a batch of items that share the same GCS config. Returns: (success_count, error_count) )rrrrBNvertex_instancerhrservice_account_jsonri-rlheadersr object_namerOz8GCS Bucket error logging batch payload to GCS bucket: %s)get_gcs_logging_configconstruct_request_headers_get_object_date_from_datetimernowrutcr'timer uuid4hexr_r_log_json_data_on_gcsrUrKr rL)r9rsrj first_kwargsgcs_logging_configrr current_dater\rcombined_payload success_count error_countrPs r;_send_grouped_batchz#GCSBucketLogger._send_grouped_batchsz#Ahx0  0@D@[@[\h@i:i #'#A#A 23D E%78N%O$B$G"4M!BK"&"E"EhllS[S_S_F`"aL!$TYY[4%7!8 94::<;K;KBQ;O:PQH!%!A!A,PX!YK&*&F&Fu&M ,,'' 0 -   JMK!;/ /-;j  0Me*K  $ $%_ab c!;/ /  0si EDD DDB:D?DDEDDD E$&E E EEEcPK|D]}|j|d{y7w)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)r9rsrvs r;_send_individual_logsz%GCSBucketLogger._send_individual_logss.  3D,,T2 2 2 3 2s &$&rvcpK |j|dd{}|j|d|dd{}|d}|j|d|d|d }|j||||d d{y7h7I7 #t$r }t j d |Yd}~yd}~wwxYww) zH Send a single log item to GCS as an individual object. rBNrrhrrrArC)rBrOrCrz=GCS Bucket error logging individual payload to GCS bucket: %s)rr_get_object_namerrKr rL)r9rvrrrrrPs r;rz%GCSBucketLogger._send_single_log_items i@D@[@[\`ai\j@k:k #'#A#A 23D E%78N%O$B$G"4M!BK!%!6!6H~ $Y!.1"7"K ,,'' $Y -  ;l   i  $ $%dfg h h isaB6B B B BAB >B?B B6B B B B3B.)B6.B33B6c K|j}|sy|jrD|j|}|jD]\}}|j ||d{!y|j |d{y7!7w)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)rZr/rwrsrr)r9rY grouped_itemsrj group_itemss r;async_send_batchz GCSBucketLogger.async_send_batchs#'"9"9";   # ##'#>#>?O#PM+8+>+>+@ H' K..{JGGG H,,-=> > >H >s$ABBB:B;BBrOrCc|jtjtj}|j dd|j |}n#|j||j dd}|j ddxsi}|j ddxsi}d |vrDt|j d d }|r&|d tjjd |}|S) zD Get the object name to use for the current payload error_strN)request_date_stridr response_idlitellm_paramsmetadata gcs_log_id)fallbackz/custom-r) rrrrrrF_generate_failure_object_name_generate_success_object_namerr rr) r9rBrOrCrr_litellm_params _metadata safe_log_ids r;rz GCSBucketLogger._get_object_names#AA(,,x||B\]   {D 1 =<>D>Q"@@%- *A "&!9!9+!FF'&*jj&:O**(  G  $$%NPXZ[\ s@4C9B0B1B CB B?B:5C:B??Crrc|d|S)N/r^)r9rrs r;rz-GCSBucketLogger._generate_success_object_nameVs ##1[M22r<cH|dtjjS)Nz /failure-)r rr)r9rs r;rz-GCSBucketLogger._generate_failure_object_name]s###9TZZ\-=-=,>??r<rc$|jdS)Nz%Y-%m-%d)strftime)r9rs r;rz.GCSBucketLogger._get_object_date_from_datetimecs$$Z00r<crK|jd{tj|_y7w)zB Override flush_queue to work with asyncio.Queue. N)rrlast_flush_timer9s r;rHzGCSBucketLogger.flush_queuefs-##%%%#yy{ &s 757cK tj|jd{tjd|j|j d{`7=7w)zE Override periodic_flush to work with asyncio.Queue. Nz*GCS Bucket periodic flush after %s seconds)r0sleepr!r rErHrs r;r5zGCSBucketLogger.periodic_flushmsX-- 3 34 4 4  !MtObOb c""$ $ $ 4 $s!$A(A$7A(A&A(&A(c Ktdw)Nz(GCS Bucket does not support health check)NotImplementedErrorrs r;async_health_checkz"GCSBucketLogger.async_health_checkvs!"LMMs )N)#__name__ __module__ __qualname__r,r&rQrSlistrJrZr_dictrrirrrwrtupler'rrrrrrrrrrrrHr5rr __classcell__)r:s@r;rrsC$J$4H0H& D$9 "5C535357d38n77(4s4s4 D,A d3PTUdPeKeFf   o1F  3  )0tO/D)0RU)0Z_`ceh`hZi)0V3o1F343iiDi8?,t>Tdglo2##!4#o #  #J333 3@@ @ 1811+%N*FNr<r)"r0rmr{r(rrrrtypingrrrlitellm._loggingr litellm._uuidr litellm.constantsr -litellm.integrations.additional_logging_utilsr /litellm.integrations.gcs_bucket.gcs_bucket_baser 1litellm.litellm_core_utils.cloud_storage_securityrlitellm.proxy._typesr,litellm.types.integrations.base_health_checkr%litellm.types.integrations.gcs_bucketlitellm.types.utilsr&litellm.llms.vertex_ai.vertex_llm_baserrr^r<r;rs` 22,,+;PI3U36AJ[Nm%;[Nr<