ZThp[ddlmZddlZddlmZddlmZddlmZmZm Z m Z m Z m Z m Z ddlZddlmZddlmZddlmZmZdd lmZdd lmZdd lmZdd lmZdd lmZddlm Z ddl!m"Z"m#Z#m$Z$m%Z%ddl&m'Z'm(Z(ddl)mZm*Z*m+Z+m,Z,m-Z-ddl.m/Z/ddl0m1Z1m2Z2m3Z3ddl4m5Z5ddl6m7Z7ejpe9Z:GddZ;y)) annotationsN)Callable)datetime)AnyDictListOptionalSetUnioncast) Timestamp) StatusCode) IngestWithConfigDataChannelValue!IngestWithConfigDataStreamRequest)IngestServiceStub) ChannelConfig)IngestionConfig)ProtobufMaxSizeExceededError) SiftChannel)IngestionValidationError)create_flow_configscreate_ingestion_config"get_ingestion_config_by_client_keyget_ingestion_config_flows) create_runget_run_id_by_name)r ChannelValue channel_fqn empty_value is_data_type)TelemetryConfig)Flow FlowConfigFlowOrderedChannelValues) RuleConfig) RuleServiceceZdZUded<ded<ded<ded<d ed <d ed <d ed<d ed<ded<ded<ded<ded<ded< d- d.dZd/dZd0dZd1dZ d2 d3dZdZ d4d Z d5d!Z d6d"Z d6d#Z d7d$Z d8d%Zd9d&Zd:d'Zd9d(Zd:d)Zd/d*Ze d;d+Ze d< d=d,Zy)>_IngestionServiceImplrtransport_channelringestion_configstr asset_namezDict[str, FlowConfig]flow_configs_by_namezSet[str]flow_configs_createdzList[RuleConfig]rules Optional[str]run_idorganization_idboolend_stream_on_errorr!configuse_lazy_flow_creationringest_service_stubr& rule_serviceNc0 |jj|||}||_||_|j r`|jsT|j Dcic]}|j|c}|_|j Dchc]}|jc}|_ n|jr:|j Dcic]}|j|c}|_t|_ nut||jDcgc]}tj|} }| Dcic]}|j|c}|_| Dchc]}|jc}|_ t||_|j"rt|j"D]@} |j$| j&vs| j&j)|j$B|j j+|j"|j"|_|j$|_||_||_|j0|_||_t5||_||_y#t$r)|jj||d}d|_YOwxYwcc}wcc}wcc}wcc}wcc}wcc}w)N) lazy_flowsT) __class___get_or_create_ingestion_configr6rr*"_ingestion_client_key_is_generatedflowsnamer-r.setringestion_config_idr#from_pbr&r8r/r, asset_namesappendcreate_or_update_rulesr)r1r2r4rr7r5) selfchannelr5r1r4force_lazy_flow_creationr*flowfr>rules g/home/www/backend.miabetepe.com/venv/lib/python3.12/site-packages/sift_py/ingestion/_internal/ingest.py__init__z_IngestionServiceImpl.__init__<sN C#~~MM,D N  +CD ' 0  4 4T=X=XFL\\(RTD(RD %?E||(Lt(LD %**IO,VTYY_,V)69e)8(<<&&q)JO,OTYY_,O)CH,I4TYY,I)'0 <<  ?$$D,<,<<$$++F,=,=> ?    4 4V\\ B\\  ++!( %55#6 #4W#=  i, /#~~MMD N  +/D '  /)S(L-W-P,Is5II:>I?4J8J J:J.I76I7c|jr|j||jjt |y)z) Perform data ingestion. N)r6_lazy_flow_creationr7IngestWithConfigDataStreamiter)rFrequestss rLingestz_IngestionServiceImpl.ingest~s5  & & $D $ $h /   ;;DNKcg}|D]5}|d}|d}|d}|j|||}|j|7|jr|j||jj t |y)z Combines the requests creation step and ingestion into a single call. See `create_ingestion_request` for information about how client-side validations are handled. flow_name timestampchannel_valuesN)create_ingestion_requestrDr6rOr7rPrQrFr>rRrIrVrWrXreqs rL ingest_flowsz"_IngestionServiceImpl.ingest_flowss  !D[)I[)I!"23N// 9nUC OOC  !  & & $D $ $h /   ;;DNKrTcg}|D]5}|d}|d}|d}|j|||}|j|7|jr|j||jj t |y)z Combines the requests creation step and ingestion into a single call. See `try_create_ingestion_request` for information about how client-side validations are handled. rVrWrXN)try_create_ingestion_requestrDr6rOr7rPrQrZs rLtry_ingest_flowsz&_IngestionServiceImpl.try_ingest_flowss  !D[)I[)I!"23N33Iy.YC OOC  !  & & $D $ $h /   ;;DNKrTcx|st||}|||_yt|||xsd|xsd|xsg||_y)z Retrieve an existing run or create one to use during this period of ingestion. Include `force_new=True` to force the creation of a new run, which will allow creation of a new run using an existing name. N)rGrun_name descriptionr2tagsmetadata)rr1r) rFrGrbrcr2rdre force_newr1s rL attach_runz _IngestionServiceImpl.attach_runsO':F!$  #)r+1r   rTcd|_y)z Detach run from this period of ingestion. Subsequent data ingested won't be associated with the run being detached. N)r1)rFs rL detach_runz _IngestionServiceImpl.detach_runs  rTc g}t|t|jk7r.tdt|jdt|dtt|D]}||}t |j }t|dk7rt d||d} |j|} | j|| } t| | jr|j| |S#t $rGtd|j|jd|j|jd |d |d wxYw) Nz Expected z channel values, got .z2Expected exactly one key in flow value, got keys: rExpected value for ` ` to be a 'z'. Instead found z in flow ) lenchannelsrrangelistkeys ValueErrortry_value_fromr data_typerDr?) rFrV flow_configrWrXvaluesi channel_dict channel_typechannel_type_keychannel_config chan_values rL+try_create_ingestion_request_ordered_valueszA_IngestionServiceImpl.try_create_ingestion_request_ordered_valuess:< ~ #k&:&:"; ;*C 4 4566KCP^L_K``ab s>*+ A)!,L 1 1 34L< A% HW ,A (11!4N +::zA_IngestionServiceImpl._is_ingest_channel_value..7s  s)doublestringint32int64r3)rrranyrs `rL_is_ingest_channel_valuez._IngestionServiceImpl._is_ingest_channel_value0s@ %!AB  ud #     rTc Z|jj|}|td|d|s td|d}g}|j|r*|j |||t t t|}nS|j|r*|j|||t t t|}ntdt|d|jdk7rtd|jd t}|j||j!|||S) a Creates an ingestion request for a flow that must exist in `flow_configs_by_name`. This method performs a series of client-side validations and will return a `IngestionValidationError` if any validations fail. Channel values can be provided as a list of `sift_py.ingestion.channel.ChannelValue` or a list of values from a `sift_py.ingestion.flow.FlowOrderedChannelValues`. zA flow config of name 'z' could not be found.z#Channel values list cannot be emptyrzUnknown channel values format: zB. Expected either ChannelValue or IngestWithConfigDataChannelValueUTCz/Expected 'timestamp' to be in UTC but it is in rk)r-rrrrr rrrrrrttypetznamer FromDatetimerY)rFrVrWrXrw first_valuerx timestamp_pbs rLr^z2_IngestionServiceImpl.try_create_ingestion_requestCsR//33I>  *))4IJ *+PQ Q$Q' 9;  ! !+ .EE; 4\8JN3[F * *; 7EET:;^L F1${2C1DESS      &*A)BRBRBTAUUVW !{ !!),,,Y 6JJrTc t}|j|t|jj||||j xsd|j xsd|jS)z Creates an ingestion request for a flow that must exist in `flow_configs_by_name`. This method does not do any sort of client-side validation and is recommended to use if performance is required. ra)rArIrWrXr1r2end_stream_on_validation_error)r rrr*rAr1r2r4)rFrVrWrXrs rLrYz._IngestionServiceImpl.create_ingestion_requestxsa!{ !!),0 $ 5 5 I I");;$" 006B+/+C+C  rTc|jjr td|D]3}|j|jvstd|jdt |j |jj||D]@}||j|j<|jj|jBy)z Tries to create a new flow at runtime. Will raise an `IngestionValidationError` if there already exists a flow with the name of the `flow_config` argument. Telemetry configs with generated ingestion client keys can not be updated at runtime.Use a custom ingestion client key if you want to update flows at runtime.z#There is already a flow with name 'rN) r5r=rr?r-rr)r*rAr.addrFrwfcs rLtry_create_flowz%_IngestionServiceImpl.try_create_flows ;; 9 9*\   bBww$333.1TUWU\U\T]]_/`aa b   " "  ! ! 5 5   3B13D % %bgg .  % % ) )"'' 2 3rTc |j|S)z) See `try_create_flows`. )rrF flow_configss rLtry_create_flowsz&_IngestionServiceImpl.try_create_flowss$t##\22rTc&|jjr tdt|j|j j ||D]@}||j|j<|jj|jBy)z Like `try_create_flow` but will not do any client side validation and raise `IngestionValidationError`. rN) r5r=rrr)r*rAr-r?r.rrs rL create_flowz!_IngestionServiceImpl.create_flows ;; 9 9*\    " "  ! ! 5 5   3B13D % %bgg .  % % ) )"'' 2 3rTc |j|S)z$ See `create_flow`. )rrs rL create_flowsz"_IngestionServiceImpl.create_flowss t..rTcjt}|D]O}|j|jvs|j|jvs5|j |jQ|Dcgc]}|j|}}|rY t |j |jj||D]'}|jj |j )yycc}w#ttjf$r}t|tjr"|jtjk7r|D]q} t |j |jj|g1#tj$r,}|jtjk7rYd}~kd}~wwxYwYd}~d}~wwxYw)z Used for lazy flow creation, which registers flows with sift as they are seen, instead of during the service initialization N)r@rIr.r-rrr)r*rArgrpcRpcErrorrcoderALREADY_EXISTSr?)rFrRmissing_flow_config_namesrequestflow_config_nameflow_configs_to_createerws rLrOz)_IngestionServiceImpl._lazy_flow_creations /2e! "    % %&6 7" " " "#**))==*, 6 @ ))--k.>.>? @1 " " 1$--@ "a/vvx:#<#<<$: "K"+ 22 11EE(M  =="668z'@'@@!A" " "sI#C>+CF25AF-7,E$#F-$F#7"FF-F##F--F2c|j}t|dk(ryt||}t|Dcic]\}}|j|}}}g}d} d} |D]{} |j | j} | |j | 2|| } | jDchc] }| | }}| jD]}| ||vstd}t|dkDrt||yycc}}wcc}w)z Compares local flows from a telemetry config with the flows registered in Sift. If a local flow contains channels that isn't in Sift then this will fail. rNc6t|d|jSNrk)rrvxs rLz<_IngestionServiceImpl._update_flow_configs.. sQ(!++7rTcJt|d|jjSr)rrvrrs rLrz<_IngestionServiceImpl._update_flow_configs..s!Q(!++*;*;)<=rTz3Encountered duplicate flow with mismatched channels) r>ror enumerater?rrDrprr)rGrAtelemetry_config config_flows sift_flowsryrIsift_flow_indices_by_nameflows_to_createsift_channel_identifierconfig_channel_identifier config_flowsift_flow_index sift_flowsift_channel_identifiersconfig_channels rL_update_flow_configsz*_IngestionServiceImpl._update_flow_configssE(-- |  ! /9LM AJ:AV$WgaTYY\$W!$W,. 8  > "( K7;;Krr,r2)clsrGr5r:r*rs rLr<z5_IngestionServiceImpl._get_or_create_ingestion_config.s>gvGbGbc  '88J''((2B2V2VX^_'' L!<r$)r>r")NNNNF)rGrrbr+rcr0r2r0rdzOptional[List[str]]rez,Optional[Dict[str, Union[str, float, bool]]]rfr3) rVr+rwr#rWrrX&List[IngestWithConfigDataChannelValue]returnr) rVr+rwr#rWrrXzList[ChannelValue]rr)rrrr3)rVr+rWrrXzAUnion[List[ChannelValue], List[IngestWithConfigDataChannelValue]]rr)rVr+rWrrXrrr)rwr#)rr#)rGrrAr+rr!)F)rGrr5r!r:r3rr)__name__ __module__ __qualname____annotations__rMrSr\r_rgrirrrrr^rYrrrrrO staticmethodr classmethodr<rrTrLr(r(,s^""%%O//""  ""   ** !%$)). @@ @ @ " @ #' @DLL(L0&*)-$(AE   #  '  "  ?  >!! ! ! ? ! 0 !F** * * + * 0 *X   &3K3K3KZ 3K + 3Kj   ?  +  .323 3(/ ,@\5O5O365OJY5O5OnOT  !  +:  HL      rTr()< __future__rloggingcollections.abcrrtypingrrrr r r r rgoogle.protobuf.timestamp_pb2r rsift.ingest.v1.ingest_pb2rrsift.ingest.v1.ingest_pb2_grpcr/sift.ingestion_configs.v2.ingestion_configs_pb2rChannelConfigPbr sift_py.errorrsift_py.grpc.transportr!sift_py.ingestion._internal.errorr,sift_py.ingestion._internal.ingestion_configrrrrsift_py.ingestion._internal.runrrsift_py.ingestion.channelrrrr "sift_py.ingestion.config.telemetryr!sift_py.ingestion.flowr"r#r$sift_py.ingestion.rule.configr%sift_py.rule.servicer& getLoggerrloggerr(rrTrLrs"$>>> 3=\K6.F K?MM4,   8 $c c rT