
    oi;                     x   d dl Z d dlZd dlZd dlZd dlZd dlZd dlZd dlZd dlZd dl	Z	d dl
Z
d dlmZ d dlmZ d dlZd dlZd dlZd dlZd dlZ G d d      Zd Zd Zd Z G d	 d
ej2                        Z G d de      Z G d de      Z G d de      Z G d de      Z G d de      Z G d de      Z ddZ!y)    N)
namedtuple)update_wrapperc                   z    e Zd Zi Zi Zed        Zd Zed        Zed        Z		 	 	 ddZ
d ZddZd	 Z	 	 	 dd
Zy)CmfDeferredJobWrapperc                 8    | j                    d| j                   S )N.)
__module____name__funcs    ./cmf/cmf_deferred_job.pygen_task_namez#CmfDeferredJobWrapper.gen_task_name   s    
 //"!DMM?33    c                 N    | j                   j                   d| j                   dS )Nz(name=))	__class__r
   nameselfs    r   __repr__zCmfDeferredJobWrapper.__repr__"   s$    ..))*&1==r   c                     | dk(  rd} | S )Nz	@minutelyz* * * * * H )schedules    r   adapt_schedulez$CmfDeferredJobWrapper.adapt_schedule%   s    {"(r   c                     	 t         j                   j                  | d       y # t         j                  $ r}t        d|        |d }~ww xY w)N   1hash_idzinvalid schedule: )croniterexpandCroniterBadCronError
ValueError)r   es     r   validate_schedulez'CmfDeferredJobWrapper.validate_schedule+   sN    	E$$Xt$<,, 	E1(<=1D	Es   !$ AAANc                    t        | |       || _        | j                  |      | _        |	xs | j                  | _        || _        || _        || _        || _        || _	        || _
        || _        |
| _        | j                  rz|st        d| j                   d      |st        d| j                   d      	 | j                  | j                  | j                               | | j                   | j                  <   | | j"                  | j                  <   y # t        $ r+}t        d| j                   d| j                         |d }~ww xY w)Nz$CmfDeferredJobWrapper: periodic job(z ) must be with option only_once.z!) must be with option system_job.zinvalid job configuration(z): )r   r   r   r   descriptionpriority
system_job	only_onceonly_once_argssoft_time_limit	countdownr   show_bg_progressbarr"   r$   r   periodicregistry)r   r   r'   r)   r*   r+   r(   r,   r   r&   r-   r#   s               r   __init__zCmfDeferredJobWrapper.__init__2   s5    	tT"	&&t,	&3$)) $",." #6 == #G		{Rr!stt #G		{Rs!tuud&&t':':4=='IJ (,DMM$))$#'dii   d #=dii[DMM?![\bccds   :*D 	E &EEc                 &     | j                   |i |S Nr   r   argskwargss      r   __call__zCmfDeferredJobWrapper.__call__O   s    tyy$)&))r   c                 6    |g }|i } | j                   |i |S r2   r   r3   s      r   applyzCmfDeferredJobWrapper.applyR   s,    <D>Ftyy$)&))r   c                 (    | j                  ||      S )N)r4   r5   )apply_asyncr3   s      r   delayzCmfDeferredJobWrapper.delayY   s    T&99r   c                    ddl m}m} dd l}|g }|i }|| j                  }|| j
                  }|| j                  }|| j                  }|| j                  }|j                  | j                  | j                  ||      }|| j                  |_        |	s|j                  j                  }	|s|	|_        || j                  }|r&t!               t#        j$                  |      z   |_        | j                  rt)        | j                        }d}| j*                  X|r*t-        d| j                   d|       f |t)        |      z  }|rL|t)        t/        |      D ci c]  }|||   
 c}      z  }n"| j*                  D ]  }|t)        ||         z  } |r7|dt1        j2                  |j5                               j7                         z   z  }||_        |j                  j;                  |d	d
g      }|r|j<                  j>                  jA                  |j                  |j                  dd       |jC                  d| j                   d       |j&                  r|j&                  r|j&                  |j&                  k  rz|jC                  d| j                   d|j                          |j&                  |_        |j<                  jD                  jG                         5  |jI                          d d d        y y |
|jJ                  }
||jL                  }||jN                  }||| j                  | j*                  |j                  jP                  | j
                  ||j                  jP                  ||
|||dd|_)        |j<                  jD                  jG                         5  |jI                          d d d        |rU|j                  jU                  t)        |j                               ddl m+}  |ddt)        |j                        i|	g       t)        |j                        S c c}w # 1 sw Y   y xY w# 1 sw Y   xY w)Nr   modelsg)r   r)   r&   r-   seconds zJCmfDeferredJobWrapper.apply_async: !!! Warning! Use args + only_once! job=z, args=__open*)only_once_keystatusfieldsskip_sched_only_once   zCmfDeferredJobWrapper(z): already queuedz): reschedule existent job )r)   r*   	person_idr(   r+   r'   r,   
admin_modecomponent_idsession_tab_idr-   )r4   r5   options)cmf_emit_eventz deferred-job-show_bg_progressbarjob_id)event_persons),cmf.includer>   r?   
cmf.fieldsr+   r(   r'   r&   r-   CmfDeferredJobr   r)   current_personidrK   r,   cmf_nowdatetime	timedeltaplan_start_datetimestrr*   printsortedhashlibmd5encode	hexdigestrF   getutilmetricswrite_deferred_job_counterdebugcmfutildisable_aclsaveacl_admin_moderM   rN   valueparams_jsonappendrP   )r   r4   r5   r'   r+   r(   r,   r&   r-   g_current_user_idg_acl_admin_modeg_component_idg_session_tab_idr>   r?   cmfjobrF   args_strkarg_keyexistent_jobrP   s                          r   r:   z!CmfDeferredJobWrapper.apply_async\   s    	*<D>F""22OJ}}H**K&"&":":##dnn+cv $ x==CL  ! 0 0 3 3-CM I&-i(2D2DY2W&WC#>>		NMH""*
 66:ii[vOP Q D	)H6&>$JaQq	\$J KKH  $22 5GF7O 44H5 HOO4E(F(P(P(R!RR -C "0044=Y_ilhm4nL  ;;CLL#((Tjlmn0;LMN33!$!8!8C<S<SVbVvVv<vGG4TYYK?Z[g[j[jZklm7:7N7NL4))557 ,$))+,  # //!^^N# // !^^"&"5"5 ]]00"oo#2LL..&. ."2':
$ XX))+ 	HHJ	 !!((SVV52=3svv;@Whygz{366{ %K*, >	 	s   P1
&P6,Q6P?Q)	NFNNFNNNF)NN)NNNNNNNNNNNN)r
   r	   __qualname__r/   r.   staticmethodr   r   r   r$   r0   r6   r8   r;   r:   r   r   r   r   r      s    HH4 4>  
 E E HL_c %(:**: sw:>imvr   r   c                  t    fd}d}| r#t        |       dk(  r| d   }nt        d|  d      |r ||      S |S )za
    job decorator:
        return object like celery task with __call__, delay, apply_async
    c                     t        | fi S r2   )r   )func_r5   s    r   wrapperz!cmf_deferred_job.<locals>.wrapper   s    $U5f55r   NrJ   r   z+cmf_deferred_job takes exactly 1 argument (z given)len	TypeError)r4   r5   r~   r   s    `  r   cmf_deferred_jobr      sM    
6 Dt9>7DI$vVWWt}r   c                  f    dd l } | j                  j                  j                         j                  S )Nr   )rT   rH   CmfDateTimenowrl   )rs   s    r   rX   rX      s$    ::!!%%'---r   c                     t        j                         }| d   }t        j                  | d   | d   |d|       |j	                         }|j                          |dd dk(  r|dd }|S )zfrom logging Formatter class   r   rJ   N
)ioStringIO	tracebackprint_exceptiongetvalueclose)eisiotbss       r   format_exceptionr      sg    
++-C	AB beRUBc:AIIKv~crFHr   c                   6     e Zd Zed        Zd fd	Zd Z xZS )JobWorkerGreenletc                  p    t         j                   j                  t         j                  j                        S r2   )rY   r   timezoneutcr   r   r   r   zJobWorkerGreenlet.now   s%      $$X%6%6%:%:;;r   c                 0    || _         t        | 	          y r2   )	worker_idsuperr0   )r   r   r   s     r   r0   zJobWorkerGreenlet.__init__  s    "r   c                 N    | j                   j                   d| j                   dS )N(r   )r   r
   r   r   s    r   r   zJobWorkerGreenlet.__repr__  s%    ..))*!DNN+;1==r   r2   )r
   r	   ry   rz   r   r0   r   __classcell__r   s   @r   r   r      s    < <>r   r   c                   V     e Zd ZdZddef fdZd Zd Zd ZddZ	d Z
d	 Zd
 Z xZS )QueueProcessorzJobQueueProcessor:get:lockr'   c                 2    || _         t        |   di | y )Nr   )r'   r   r0   )r   r'   r5   r   s      r   r0   zQueueProcessor.__init__  s     "6"r   c                    ddl m}m}m} dd l}|j
                  j                         5  	 |j                  j                  j                  | j                  dd      5  d }| j                  dd| j                  g}g dd	d| j                  gg}|r|j                  |       |j                  j                  |
      D ]:  }|j!                  d       t#        j$                  dd| d| j                   d       < g ddg dddt'               ggg}	|r|	j                  |       |j                  j)                  g dg dg dgddg      }
|
r1|	j                  dg ddd|
D cg c]  }|j*                   c}gg       |j                  j-                  |	dgd|      }|r]d|_        |j0                  j3                          | j                  |_        |j5                  d       |j7                          	 d d d        n]	 d d d         |        | j                  st9        j:                  d        n)t9        j:                  t=        d | j                               /	 d d d        |S c c}w # 1 sw Y   mxY w# 1 sw Y   S xY w)!Nr   )r>   socketio
cmf_commitT
   timeoutblocking_timeoutr'   =rG   r   in_progressr   filterz+Job in_progress but queue processor is not.z%sz!!! Not running job , priority=z, mark as dead!)rG   r   rD   OR)r[   r   Nr[   z<=)r)   r   T)rF   z!=Nz--rF   )r   rH   )rF   r   NNOT INcmf_created_at)r   order_by
for_updaterH   r   z	job startr   )rS   r>   r   r   cmf.appappcmf_contextrd   rh   CmfLock_GET_LOCK_KEYr'   r   rn   rU   liston_errorloggingerrorrX   slistrF   rc   rG   start_datetimeset_now	error_logrj   timesleepmax)r   rH   r>   r   r   rs   priority_filterstuck_filter	stuck_job
job_filteronly_once_in_progressjob_next_jobs                r   _job_get_nextzQueueProcessor._job_get_next  s   <<WW  " :	2!!))$*<*<b[])^ 1"&==,'13&FO 3 #t~~6  # ''8!'!6!6!;!;<!;!P uI&&'TUMM$*>ykUYUbUbTccr(stu
 ,:.gi@
 #%%o6(.(=(=(C(C46SUst /2 )D )4% )%% 8,hXm8nPT9K9K8no "0044%1A0Bt! 5 # &3HO++335)-H&&&4MMOc1 1V W1d L==

1 

3q$--01s 1:	2z - 9oK1 1:	2z s>   4I6DI*"I%5A?I*4	I6>AI6%I**I3	/I66J c           
         ddl m} dd l}|j                  j	                         5  |j
                  j                  j                  |j                  |j                  dd       |j
                  j                  j                  |j                  |j                  dt        t        j                  j                         |j                  z
  j                         dz               |j!                  |       d d d        y # 1 sw Y   y xY w)Nr   )r   successrJ   duration  )rS   r   r   r   r   rd   re   rf   r'   r   write_deferred_job_timingsintrY   r   r   total_seconds
on_success)r   rt   resultr   rs   s        r   _job_set_successzQueueProcessor._job_set_successV  s    (WW  " 	#HH77chhPY[\]HH77chhPZ8;X=N=N=R=R=TWZWiWi=i<x<x<z  ~B  =B  9CDNN6"		# 	# 	#s   CC;;Dc           
      P   dd l }|j                  j                         5  |j                  j                  j                  |j                  |j                  dd       |j                  j                  j                  |j                  |j                  dt        t        j                  j                         |j                  z
  j                         dz               d}t        j                         }|d   rt!        |      }|j#                  ||       d d d        y # 1 sw Y   y xY w)Nr   r   rJ   r   r   zError?)r   r   r   rd   re   rf   r'   r   r   r   rY   r   r   r   sysexc_infor   r   )r   rt   r   rs   
error_textr   s         r   _job_set_failzQueueProcessor._job_set_failb  s    WW  " 		/HH77chhPWYZ[HH77chhPZ8;X=N=N=R=R=TWZWiWi=i<x<x<z  ~B  =B  9CD "J||~H{-h7
LLX.		/ 		/ 		/s   C4DD%c                 ^    ddl mm dd lfd}t	        j
                  ||      S )Nr   r=   c            
         t         j                  j                     } j                  j	                         5  j
                  j                  j                  j                  j                  dd       j                  dkD  r<j
                  j                  j                  j                  j                  dd       j                  }|sj                  }t        j                  j                         |z
  j                         }j
                  j                  j                  j                  j                  dt!        |dz               j"                  j%                         j&                  rAj&                  j(                  j*                  k7  rj                  j-                  j.                  j1                  j&                  j                  j2                  j4                               j6                  j9                          j:                  d   j1                  d	d
      rj6                  j=                          j:                  d   j1                  dd
      rj:                  d   d   _        j:                  d   j1                  dd
      rj:                  d   d   _          | j:                  d   i j:                  d   cd d d        S # 1 sw Y   y xY w)NrunrJ   r   	run_retrystart_latencyr   rH   rO   rL   FrM   rN   r4   r5   )!r   r/   r   r   r   rd   re   rf   r'   retry_countr[   r   rY   r   r   r   r   rU   set_current_deferred_jobrK   system_personrW   set_current_person	CmfPersonrc   APPcurrent_person_fieldsCmfAccessListsetup_contextrm   activate_admin_moderM   rN   )job_funcrun_latency_planrun_latency_secrs   r?   rt   r>   s      r   runnerz'QueueProcessor._job_run.<locals>.runnerv  s;   ,55chh?H$$& W  ;;CLL#((TY[\]??Q&HH$$??chhXcefg#&#:#: ''*'9'9$#+#4#4#8#8#:=M#M"\"\"^  ;;CLL#((Tc<?RV@V<WY %%>>sC==S]]aoo6H6H%HGG..((,,S]]377;;CdCd,eg((668y155lEJ,,@@By155neL),)CN)Sy1556FN+.??9+EFV+W(!8VCOOH<UV5W W Ws   J#K&&K/)rS   r>   r?   r   geventwith_timeout)r   rt   r+   r   rs   r?   r>   s    `  @@@r   _job_runzQueueProcessor._job_runr  s$    )	W< ""?F;;r   c           	         | j                  dg      }|s7t        j                  j                  j	                  | j
                  ddd       y |j                  d   j                  d      }	 | j                  ||      }| j                  ||       y # t        t        t        j                  f$ rS}t        j                  d|j                    d	|j"                   d
|        | j%                  |t'        |              d }~wt(        t        j*                  f$ rU}t        j                  d|j                    d	|j"                   d       | j%                  |t'        |             Y d }~y d }~ww xY w)NrE   r   idlerJ   rO   r+   )r+   zjob_runner(z, z) interrupt by ) error)r   rs   rd   re   rf   r'   rm   rc   r   r   
SystemExitKeyboardInterruptr   GreenletExitr   	exceptionrW   r   r   r\   	ExceptionTimeout)r   rt   r+   r   r#   s        r   
_run_cyclezQueueProcessor._run_cycle  s/      .HH77vvWXY//)4889JK	,]]3]HF !!#v.-v/B/BC 	CFF82chhZqcRSsCF+6>>* 	,CFF82chhZwGHsCF++	,s&   ,%B E/0AC>>E/AE**E/c                    	 	 | j                          t        j                  d       '# t        t        t
        j                  f$ r  t        $ r0 t        j                  |  d       t        j                  d       Y Ww xY w)zmain() for GreenletrJ   z cycle error.   )
r   r   r   r   r   r   r   r   r   r   r   s    r   _runzQueueProcessor._run  sn    !

1	 
  163F3FG  !!TF-"89

2s   %) AB ?B c                 h    | j                   j                   d| j                   d| j                   dS )Nr   r   r   )r   r
   r   r'   r   s    r   r   zQueueProcessor.__repr__  s0    ..))*!DNN+;;t}}oUVWWr   r2   )r
   r	   ry   r   r   r0   r   r   r   r   r   r   r   r   r   s   @r   r   r     s;    0M# #AF#/ "<H,.Xr   r   c                       e Zd ZdZd Zy)QueueCleanerzJobQueueCleaner:lockc           
      h   dd l }ddlm}m} 	 	 |j                  j
                  j                  | j                  dd      5  |j                  j                         5  t               }|t        j                  d      z
  }|t        j                  d      z
  }|j                  j                  d	d
dg dgdd|ggg ddd|ggg       d d d        d d d        t%        j&                  d       # 1 sw Y   'xY w# 1 sw Y   +xY w# |$ r Y 5t        $ r t!        j"                  |  d       Y Ww xY w)Nr   CmfGetLockErrorr>   ,  r      )days   )hoursr   rG   IN)failcanceldeadend_datetime<)rG   r   r   r    error)r   rS   r  r>   rd   rh   r   	_LOCK_KEYr   r   rX   rY   rZ   rU   bulk_deleter   r   r   r   r   )r   rs   r  r>   r   max_end_datetimemax_success_end_datetimes          r   r   zQueueCleaner._run  s.   73XX%%--dnnc\]-^ 
`c`g`g`s`s`u 
!)C'*X-?-?Q-G'G$/2X5G5Gb5Q/Q,))55 &.HIN\_aqKrs7.#Og9hi  6 
 
 JJsO% 
 
 
 
 #  3!!TF&/23sH   2D C;A-C/	C;D /C8	4C;;D D D1 D10D1Nr
   r	   ry   r  r   r   r   r   r  r    s    &Ir   r  c                       e Zd ZdZd Zy)QueueWatcherzJobQueueWatcher:lockc                    dd l }ddlm}m} |j                  j
                  j                  j                  }	 	 |j                  j                  j                  | j                  dd      5  |j                  j                         5  |j                  d      }|r#t        j                         t        |      z
  dk  r't        j                   d       	 d d d        d d d        |j#                  d      }t%        |      D ]u  }|j                  d	|j'                                &|j)                  d|       |j+                  |       t-        d
|j'                          dt.        j0                         w t3               }|t5        j6                  d      z
  }	|j8                  j%                  dgdd|D 
cg c]  }
|
j'                          c}
gg ddd|	gg      D ]  }|j;                  d        	 d d d        d d d        t        j                   d       c c}
w # 1 sw Y   -xY w# 1 sw Y   1xY w# |$ r Y ;t<        $ r t?        j@                  |  d       Y ]w xY w)Nr   r  r  r   redis_idr   r   deferred_job_worker:workersdeferred_job_worker:heartbeat:zQueueWatcher: worker z seems dead!filer@   rE   r   r   r   r   r  )rH   r   zJob Worker is deadr  <   )!r   rS   r  r>   r   r   REDIS_DBredisrd   rh   r   r  r   rc   r   floatr   smembersr   decodesremremover]   r   stderrrX   rY   rZ   rU   r   r   r   r   )r   rs   r  r>   redis_dbr  
db_workersr   r   max_start_datew_idrt   s               r   r   zQueueWatcher._run  s&   7"%''++"6"6"<"<3XX%%--dnnc\]-^ ;`c`g`g`s`s`u ;'||J7H#tyy{U8_'Dr'I

2 ; ; ; "*!2!25P!RJ%)*%5 m	#<<*HIYIYI[H\(]^f$MM,G)T&--i8!$9):J:J:L9M\"Zadakakl	m ")C%(8+=+=c+J%JN%4499$'5!,hS]8^48^ _ >!13 G$  :   ; %9:;#; ;: JJrNA . 9_); ; ; ;2 #  3!!TF&/23st   2H; +H/AH#H/H; $AH#&BH#8H'H#7H/?H; H##H,	(H//H84H; ;I% I%$I%Nr  r   r   r   r  r    s    &I&r   r  c                   :     e Zd ZdZ edd      Z fdZd Z xZS )JobScheduleru   
    - защита от параллельного запуска
    - "размазывание нагрузки" во времени.
    PeriodicDatazjob croniterc                 2    t        |   |i | g | _        y r2   )r   r0   r.   )r   r4   r5   r   s      r   r0   zJobScheduler.__init__  s    $)&)r   c           
         dd l }ddlm} |j                  j                  j
                  j                  }d}t        j                  j                  t        j                  j                        j                         j                  }t        j                  j                         D ]v  }|j                   }|j"                  r\|j$                  |j"                  v rD|j"                  |j$                     }t'        |  d|j$                   d|j                    d|        n(t'        |  d|j$                   d|j                    d       |st'        |  d|j$                   d       t        j)                  |      }t        j+                  |       t-        j,                  ||j.                  	      }||_        |j1                          |rt3        ||j5                               n|j5                         }| j                  j7                  | j9                  ||             y 	 |r|t;        j:                         z
  }	nt'        d       d}	|	dk  rt'        d|	dd       d}	t;        j<                  |	       	 t;        j:                         }
d}| j                  D ]  \  }}|j5                         }|
|k\  rs|j1                          |j?                  d|j$                   d| | j@                  dd
      r4	 |j                  jC                         5  |jE                          d d d        |rt3        ||j5                               n|j5                         } 	 5# 1 sw Y   :xY w# tF        $ r( tI        jJ                  |  d|j$                   d       Y mw xY w# tF        $ r0 tI        jJ                  |  d       t;        j<                  d       Y zw xY w)Nr   configz: config: job=z, app="z ", config.JOB_SCHEDULE_OVERRIDE="z: config: job z
 disabled!r   Tz1JobScheduler: warning next_ts == 0, nothing to dor   z)JobScheduler: warning too low sleep time z.3fr      zJobScheduler:lock::)exnxz schedule job(r   r  )&r   rS   r2  r   r   r   r!  rY   r   r   r   
astimezonetzinfor   r.   valuesr   JOB_SCHEDULE_OVERRIDEr   r]   r   r$   r   APP_FQDNget_nextminget_currentrn   r.  r   r   setr   r   r:   r   r   r   )r   rs   r2  r(  next_tslocal_tzinfort   r   	cron_itersleep_tsnow_tscurr_tss               r   r   zJobScheduler._run  s4   &"%''++"6"6"<"<((,,X->->-B-BCNNPWW(1188: 	DC||H++F<X<X0X!77AnSXXJgcll^KkltkuvwnSXXJgcll^1MNnSXXJjAB,;;HEH!33H= ))(FOOLI+I ?Fc'9#8#8#:;ILaLaLcGMM  !2!23	!BC#	D& "TYY[0IK1}A(3qQRJJx &*mm lNC'335G(!**,#<<*<SXXJay(QSWSaSafhmq<r\%(WW%8%8%: !6$'OO$5!6 HOc'9+@+@+BCT]TiTiTkGl *!6 !6#, \ ' 1 1TF.
RY2Z [\  !!TF&/2

2sO   3A<N 0M 
MM #.N M	M  .NN NN 6OO)	r
   r	   ry   __doc__r   r.  r0   r   r   r   s   @r   r-  r-    s      nn=L8r   r-  c                       e Zd ZdZdZd Zy)RedisCleaneruD   Раз в сутки сбрасываем редис БД. В 02-50.zRedisCleaner:lockc                    dd l }ddlm}m} |j                  j
                  j                  j                  }	 t        j                  d       	 t        j                  j                         }|j                  dk7  s|j                  dk  rT|j                  j                  j!                  | j"                  dd      5  |j%                  d      }|r#t        j                         t'        |      z
  dk  r
	 d d d        t)        d	t*        j,                  
       |j/                          d d d        # 1 sw Y   
xY w# t        j0                  j2                  $ r Y ,|$ r Y 2t4        $ r t7        j8                  |  d       Y Tw xY w)Nr   )r  	CMF_CACHEr  r   2   r   r  i`T  z%RedisCleaner: run CMF_CACHE.flushdb()r  r  )r   rS   r  rK  r   r   r   r!  r   r   rY   r   hourminuterd   rh   r   r  rc   r"  r]   r   r'  flushdb
exceptionsLockNotOwnedErrorr   r   r   )r   rs   r  rK  r(  r   r  s          r   r   zRedisCleaner._runQ  s5   :"%''++"6"6"<"<JJsO3''++-88q=CJJOXX%%--dnnc\]-^ ('||J7H#tyy{U8_'Dx'O ( ( AT%%'( ( ( ##55 "  3!!TF&/23sH   <D> 2D> =8D25D> >+D2)D> 2D;7D> >FF# FFN)r
   r	   ry   rG  r  r   r   r   r   rI  rI  M  s    N#I3r   rI  c                   f     e Zd Z fdZd ZdefdZddedefdZ	defdZ
ddedefd	Zd
 Z xZS )RedisMonitorc                     || _         |j                  | _        || _        | j                          t	        j
                  d      | _        t        |    di | y )Nzredis-monitorr   )	r(  redis_instance_nameredis_settings_manager_RedisMonitor__init_redisr   	getLoggerloggerr   r0   )r   r(  rV  r5   r   s       r   r0   zRedisMonitor.__init__n  sN     #+#?#? &<#''8"6"r   c                     ddl m} t        |d      rr|j                  | j                     d   dk(  rS|j                  | j                     j                         }|j                  d       t        j                  di || _        y t        d      )Nr   r1  cache_settingstyper!  uJ   В конфиге не содержится конфигурации Redisr   )
rS   r2  hasattrr[  rU  copypopr!  Redisr"   )r   r2  cfgs      r   __init_rediszRedisMonitor.__init_redisv  sz    & 6+,1F1FtG_G_1`ag1hls1s''(@(@AFFHCGGFO+s+DJijjr   returnc                    ddl m} t        |j                  j	                               }t        |      dk(  r|d   S 	 |j                  | j                        }||dz   t        |      z     }|S # t        $ r	 |d   }Y |S w xY w)u!  
        Возвращает следующую redis конфигурацию из config.cache_settings.
        В случае, если текущий инстанс последний в списке, то начинает обходить
        список заново
        r   r1  rJ   )	rS   r2  r   r[  keysr   indexrU  r"   )r   r2  re  idxnext_instances        r   _get_new_instancezRedisMonitor._get_new_instance  s     	'F))..01t9>7N	$**T556C #'SY!67M   	$ GM	$s   /A/ /B Bintervalc                     	 | j                         | _        | j                  j                  d| j                         | j	                          | j                         }|ryt        j                  |       t)NTu=   Получена новая конфигурация Redis %s)ri  rU  rY  warningrW  check_redis_connectionr   r   )r   rj  
is_successs      r   _check_new_instancez RedisMonitor._check_new_instance  sc    '+'='='?D$KK _aeayayz446JLL" r   c                    | j                   j                  j                  }|j                  d      xs |j                  d      }	 | j                   j	                          y# t         j
                  $ r&}| j                  j                  d|       Y d }~yd }~wt        $ r&}| j                  j                  d|       Y d }~yd }~ww xY w)NhostpathTuA   Ошибка подключения к Redis по адресу %su=   Системная ошибка с Redis по адресу %sF)	r!  connection_poolconnection_kwargsrc   pingConnectionErrorrY  rl  r   )r   rt  current_hostr#   s       r   rm  z#RedisMonitor.check_redis_connection  s     JJ66HH(,,V4U8I8M8Mf8U	oJJOO$$ 	sKK ceqrr   	oKK _amnn	os$   A! !C4BC!CCmax_failure_countc                    d}	 | j                         }|rd}n|dz  }| j                  j                  d|       ||k\  r| j                  j                  d|       | j                         }|rS| j                  j                  d| j                         | j
                  j                  | j                  d       t        t        j                  |       )Nr   TrJ   u.   Redis недоступен (%d попытка)u3   Redis недоступен %d раза подрядuA   Записываем новую конфигурацию Redis %s)
with_flush)
rm  rY  rl  ro  rU  rV  
save_to_dbr   r   r   )r   rj  rx  failure_countrn  is_new_successs         r   monitor_rediszRedisMonitor.monitor_redis  s    446J !"##$TVcd $55KK''(]_lm%)%=%=%?N%++,oqu  rJ  rJ  K33>>t?W?Wdh>i((LL"# r   c                 $    | j                          y r2   )r~  r   s    r   r   zRedisMonitor._run  s    r   )   )r  r4  )r
   r	   ry   r0   rW  r\   ri  r   boolro  rm  r~  r   r   r   s   @r   rS  rS  m  sQ    #
k3 (	#C 	# 	# #c ## #*r   rS  c                 d   ddl }ddlm} |j                  j                  }t        j                          dt        j                          }|j                  r<t        |j                  j                  j                  ||      }|j                          |j                          |j                  j                  j                  j                  }| sd|j                  j                   _        g }| r|j%                  t'        d|             n,t)        d      D ]  }|j%                  t'        ||               |j%                  t+        |	             |j%                  t-        |	             |j%                  t/        |	             |j%                  t1        |	             |D ]  }	|	j                           	 	 	 |j3                  d
| dd       |j5                  d|       t=        j>                  |d      5 }|D ]&  }	tA        d|	 dtB        jD                         tF         	 ddd       w# t6        $ r"}
t9        j:                  d|
        Y d}
~
sd}
~
ww xY w# 1 sw Y   8xY w# tF        tH        t<        jJ                  f$ r)}
tA        d|
 dtB        jD                         Y d}
~
nd}
~
ww xY w	 t=        jL                  |d       n4# t<        jN                  $ r tA        dtB        jD                         Y nw xY w|D ]  }	|	r tA        d|	 dtB        jD                         %|	jQ                         r tA        d|	 dtB        jD                         U	 |	jS                          g# tF        tH        t<        jJ                  t6        f$ r t9        j:                  d|	 d       Y w xY w y)u!  
    Функция - обработчик очередей задач.
    На каждый приоритет(очередь) запускаем отдельный поток - обработчик.
    Сервисный поток для очистки очереди от устаревших задач.
    Сервисный поток для контроля погибших задач(чей воркер погиб).
    Сервисный поток Beat, для планирования периодических задач.
    r   Nr1  r5  )r(  rV  r   T)r'   r      )r   r  -   )r6  r  z$deferred_job_worker: redis error!!! r   )r   z%deferred_job_worker(): system thread(z) exited!!! Stop work.r  zdeferred_job_worker(): stop by z. Try kill all threads...z(deferred_job_worker(): killall timeout!.zdeferred_job_worker(): Thread z is not stopped!!!z stops normally!z error.)*r   rS   r2  r   REDIS_SETTINGS_MANAGERplatformnodeuuiduuid1CACHE_REDIS_FAILOWERrS  r   r   start
init_redisr!  rK  job_daemon_modern   r   ranger  r  r-  rI  r@  saddr   r   r   r   iwaitr]   r   r'  r   r   r   killallr   
successfulrc   )single_queuers   r2  rV  r   redis_monitorr(  	greenletsr'   glr#   gl_iters               r   deferred_job_workerr    s1    " WW;;==?#1TZZ\N3I""$WW[[))#9

 	%%'!ggkk2288H,0)IKLb	 	UH^XST	U\I67\I67\I67\I67 

_N=i[I3SUV ;iH i4 %! %BA"E[\cfcmcmn$$%%   N!!$H"LMMN% % )6+>+>? _/s2KLSVS]S]^^_Ly"->> L8

KL  L22$6HIPSPZPZ[==?22$6FGcjjY	LFFH-v/B/BIN 	L >rd'JK	LLs~   <I. >)H4 'I. >,I"+	I. 4	I=II. II. "I+'I. .J5J00J59K .LLM..<N-,N-)F)"rY   r_   r   ossignalr   r   r   r   r  r  collectionsr   	functoolsr   r   r   r!  cmf.util.cmfutilrs   cmf.util.metricsr   r   rX   r   Greenletr   r   r  r  r-  rI  rS  r  r   r   r   <module>r     s      	 	    
    " $     { {|*.
> 
>oX& oXd$ :)$ )XC$ CL3$ 3@W$ WtOLr   