Commit e9dfb7d2 authored by Ondřej Kuzník's avatar Ondřej Kuzník
Browse files

ITS#9600 Rework monitor entry management

Connection entries are now persistent and can be addressed, preparing
the ground so we can process modify operations on individual entries.
parent 6d95cc39
......@@ -608,6 +608,14 @@ client_init(
/* We only register the write event when we have data pending */
event_add( c->c_read_event, c->c_read_timeout );
#ifdef BALANCER_MODULE
if ( lload_monitor_client_subsys &&
lload_monitor_conn_entry_create(
c, lload_monitor_client_subsys ) ) {
goto fail;
}
#endif /* BALANCER_MODULE */
checked_lock( &clients_mutex );
LDAP_CIRCLEQ_INSERT_TAIL( &clients, c, c_next );
checked_unlock( &clients_mutex );
......@@ -741,6 +749,17 @@ client_destroy( LloadConnection *c )
CONNECTION_LOCK(c);
assert( c->c_state == LLOAD_C_DYING );
#ifdef BALANCER_MODULE
/*
* Can't do this in client_unlink as that could be run from cn=monitor
* modify callback.
*/
if ( !BER_BVISNULL( &c->c_monitor_dn ) ) {
lload_monitor_conn_unlink( c );
}
#endif /* BALANCER_MODULE */
c->c_state = LLOAD_C_INVALID;
assert( c->c_ops == NULL );
......
......@@ -495,6 +495,10 @@ struct LloadConnection {
TAvlnode *c_linked;
#ifdef BALANCER_MODULE
struct berval c_monitor_dn;
#endif /* BALANCER_MODULE */
/*
* Protected by the CIRCLEQ mutex:
* - Client: clients_mutex
......@@ -583,12 +587,6 @@ struct LloadListener {
typedef int (*CONNCB)( LloadConnection *c, void *arg );
struct lload_monitor_conn_arg {
Operation *op;
monitor_subsys_t *ms;
Entry **ep;
};
/* config requires a bi_private with configuration data - dummy for now */
struct lload_conf_info {
int dummy;
......
......@@ -34,6 +34,7 @@
#include "lload.h"
#include "lber_pvt.h"
#include "lutil.h"
#include "ldap_rq.h"
#include "lload-config.h"
......@@ -86,12 +87,15 @@ static AttributeDescription *ad_olmRejectedOps;
static AttributeDescription *ad_olmCompletedOps;
static AttributeDescription *ad_olmFailedOps;
static AttributeDescription *ad_olmConnectionType;
static AttributeDescription *ad_olmConnectionState;
static AttributeDescription *ad_olmPendingOps;
static AttributeDescription *ad_olmPendingConnections;
static AttributeDescription *ad_olmActiveConnections;
static AttributeDescription *ad_olmIncomingConnections;
static AttributeDescription *ad_olmOutgoingConnections;
monitor_subsys_t *lload_monitor_client_subsys;
static struct {
char *name;
char *oid;
......@@ -197,6 +201,13 @@ static struct {
"NO-USER-MODIFICATION "
"USAGE dSAOperation )",
&ad_olmOutgoingConnections },
{ "( olmBalancerAttributes:13 "
"NAME ( 'olmConnectionState' ) "
"DESC 'Connection state' "
"EQUALITY caseIgnoreMatch "
"SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 "
"USAGE dSAOperation )",
&ad_olmConnectionState },
{ NULL }
};
......@@ -253,6 +264,7 @@ static struct {
"SUP top STRUCTURAL "
"MAY ( "
"olmConnectionType "
"$ olmConnectionState "
"$ olmPendingOps "
"$ olmReceivedOps "
"$ olmCompletedOps "
......@@ -519,82 +531,24 @@ done:
return rc;
}
/*
* Monitor cache is locked, the connection cannot be unlinked and freed under us.
* That also means we need to unlock and finish as soon as possible.
*/
static int
lload_monitor_in_conn_entry( LloadConnection *conn, void *argv )
{
Entry *e;
monitor_entry_t *mp;
struct lload_monitor_conn_arg *arg = argv;
monitor_extra_t *mbe = arg->op->o_bd->bd_info->bi_extra;
char buf[SLAP_TEXT_BUFLEN];
struct berval bv;
bv.bv_val = buf;
bv.bv_len = snprintf(
bv.bv_val, SLAP_TEXT_BUFLEN, "cn=Connection %lu", conn->c_connid );
e = mbe->entry_stub( &arg->ms->mss_dn, &arg->ms->mss_ndn, &bv,
oc_olmBalancerConnection, NULL, NULL );
mp = mbe->entrypriv_create();
e->e_private = mp;
mp->mp_info = arg->ms;
mp->mp_flags = MONITOR_F_SUB | MONITOR_F_VOLATILE;
*arg->ep = e;
arg->ep = &mp->mp_next;
return 0;
}
static int
lload_monitor_in_conn_create(
Operation *op,
SlapReply *rs,
struct berval *ndn,
Entry *e_parent,
Entry **ep )
{
monitor_entry_t *mp_parent;
struct lload_monitor_conn_arg arg = {
.op = op,
.ep = ep,
};
assert( e_parent->e_private != NULL );
mp_parent = e_parent->e_private;
arg.ms = (monitor_subsys_t *)mp_parent->mp_info;
checked_lock( &clients_mutex );
connections_walk(
&clients_mutex, &clients, lload_monitor_in_conn_entry, &arg );
checked_unlock( &clients_mutex );
return 0;
}
static int
lload_monitor_up_conn_entry( LloadConnection *c, void *argv )
lload_monitor_conn_update( Operation *op, SlapReply *rs, Entry *e, void *priv )
{
Entry *e;
monitor_entry_t *mp;
struct lload_monitor_conn_arg *arg = argv;
monitor_extra_t *mbe = arg->op->o_bd->bd_info->bi_extra;
char buf[SLAP_TEXT_BUFLEN];
struct berval bv_rdn,
bv_type = BER_BVNULL,
bv_pending = BER_BVNULL,
bv_received = BER_BVNULL,
bv_completed = BER_BVNULL,
bv_failed = BER_BVNULL;
Attribute *a;
LloadConnection *c = priv;
struct berval bv_type, bv_state;
ldap_pvt_mp_t active, pending, received, completed, failed;
bv_rdn.bv_val = buf;
bv_rdn.bv_len = snprintf(
bv_rdn.bv_val, SLAP_TEXT_BUFLEN, "cn=Connection %lu", c->c_connid );
CONNECTION_LOCK(c);
e = mbe->entry_stub( &arg->ms->mss_dn, &arg->ms->mss_ndn, &bv_rdn,
oc_olmBalancerConnection, NULL, NULL );
pending = (ldap_pvt_mp_t)c->c_n_ops_executing;
received = c->c_counters.lc_ops_received;
completed = c->c_counters.lc_ops_completed;
failed = c->c_counters.lc_ops_failed;
switch ( c->c_type ) {
case LLOAD_C_OPEN: {
......@@ -619,72 +573,160 @@ lload_monitor_up_conn_entry( LloadConnection *c, void *argv )
} break;
}
UI2BV( &bv_pending, (long long unsigned int)c->c_n_ops_executing );
UI2BV( &bv_received, c->c_counters.lc_ops_received );
UI2BV( &bv_completed, c->c_counters.lc_ops_completed );
UI2BV( &bv_failed, c->c_counters.lc_ops_failed );
attr_merge_normalize_one( e, ad_olmConnectionType, &bv_type, NULL );
attr_merge_normalize_one( e, ad_olmPendingOps, &bv_pending, NULL );
attr_merge_normalize_one( e, ad_olmReceivedOps, &bv_received, NULL );
attr_merge_normalize_one( e, ad_olmCompletedOps, &bv_completed, NULL );
attr_merge_normalize_one( e, ad_olmFailedOps, &bv_failed, NULL );
ch_free( bv_pending.bv_val );
ch_free( bv_received.bv_val );
ch_free( bv_completed.bv_val );
ch_free( bv_failed.bv_val );
mp = mbe->entrypriv_create();
e->e_private = mp;
mp->mp_info = arg->ms;
mp->mp_flags = MONITOR_F_SUB | MONITOR_F_VOLATILE;
*arg->ep = e;
arg->ep = &mp->mp_next;
switch ( c->c_state ) {
case LLOAD_C_INVALID: {
/* *_destroy removes the entry from list before setting c_state to
* INVALID */
assert(0);
} break;
case LLOAD_C_READY: {
struct berval bv = BER_BVC("ready");
bv_state = bv;
} break;
case LLOAD_C_CLOSING: {
struct berval bv = BER_BVC("closing");
bv_state = bv;
} break;
case LLOAD_C_ACTIVE: {
struct berval bv = BER_BVC("active");
bv_state = bv;
} break;
case LLOAD_C_BINDING: {
struct berval bv = BER_BVC("binding");
bv_state = bv;
} break;
case LLOAD_C_DYING: {
/* I guess we got it before it was unlinked? */
struct berval bv = BER_BVC("dying");
bv_state = bv;
} break;
default: {
struct berval bv = BER_BVC("unknown");
bv_state = bv;
} break;
}
CONNECTION_UNLOCK(c);
a = attr_find( e->e_attrs, ad_olmConnectionType );
assert( a != NULL );
if ( !(a->a_flags & SLAP_ATTR_DONT_FREE_DATA) ) {
ber_memfree( a->a_vals[0].bv_val );
a->a_flags |= SLAP_ATTR_DONT_FREE_DATA;
}
a->a_vals[0] = bv_type;
a = attr_find( e->e_attrs, ad_olmConnectionState );
assert( a != NULL );
if ( !(a->a_flags & SLAP_ATTR_DONT_FREE_DATA) ) {
ber_memfree( a->a_vals[0].bv_val );
a->a_flags |= SLAP_ATTR_DONT_FREE_DATA;
}
a->a_vals[0] = bv_state;
a = attr_find( e->e_attrs, ad_olmPendingOps );
assert( a != NULL );
UI2BV( &a->a_vals[0], pending );
a = attr_find( e->e_attrs, ad_olmReceivedOps );
assert( a != NULL );
UI2BV( &a->a_vals[0], received );
a = attr_find( e->e_attrs, ad_olmCompletedOps );
assert( a != NULL );
UI2BV( &a->a_vals[0], completed );
a = attr_find( e->e_attrs, ad_olmFailedOps );
assert( a != NULL );
UI2BV( &a->a_vals[0], failed );
return SLAP_CB_CONTINUE;
}
int
lload_monitor_conn_unlink( LloadConnection *c )
{
BackendInfo *mi = backend_info( "monitor" );
monitor_extra_t *mbe = mi->bi_extra;
assert( mbe && mbe->is_configured() );
CONNECTION_ASSERT_LOCKED(c);
assert( !BER_BVISNULL( &c->c_monitor_dn ) );
/*
* Avoid a lock inversion with threads holding monitor cache locks in turn
* waiting on CONNECTION_LOCK(c)
*/
CONNECTION_UNLOCK(c);
mbe->unregister_entry( &c->c_monitor_dn );
CONNECTION_LOCK(c);
ber_memfree( c->c_monitor_dn.bv_val );
BER_BVZERO( &c->c_monitor_dn );
return 0;
}
static int
lload_monitor_up_conn_create(
Operation *op,
SlapReply *rs,
struct berval *ndn,
Entry *e_parent,
Entry **ep )
int
lload_monitor_conn_entry_create( LloadConnection *c, monitor_subsys_t *ms )
{
monitor_entry_t *mp_parent;
monitor_subsys_t *ms;
LloadBackend *b;
struct lload_monitor_conn_arg arg = {
.op = op,
.ep = ep,
};
char buf[SLAP_TEXT_BUFLEN];
char timebuf[LDAP_LUTIL_GENTIME_BUFSIZE];
struct tm tm;
struct berval bv_rdn, bv_timestamp, zero = BER_BVC("0"),
value = BER_BVC("unknown");
monitor_entry_t *mp;
monitor_callback_t *cb;
Entry *e;
Attribute *a;
BackendInfo *mi = backend_info( "monitor" );
monitor_extra_t *mbe = mi->bi_extra;
assert( e_parent->e_private != NULL );
assert( mbe && mbe->is_configured() );
mp_parent = e_parent->e_private;
ms = (monitor_subsys_t *)mp_parent->mp_info;
b = ms->mss_private;
CONNECTION_ASSERT_LOCKED(c);
assert( BER_BVISNULL( &c->c_monitor_dn ) );
if ( !b ) {
return -1;
}
bv_rdn.bv_val = buf;
bv_rdn.bv_len = snprintf(
bv_rdn.bv_val, SLAP_TEXT_BUFLEN, "cn=Connection %lu", c->c_connid );
arg.ms = ms;
ldap_pvt_gmtime( &c->c_activitytime, &tm );
bv_timestamp.bv_len = lutil_gentime( timebuf, sizeof(timebuf), &tm );
bv_timestamp.bv_val = timebuf;
checked_lock( &b->b_mutex );
connections_walk_last( &b->b_mutex, &b->b_conns, b->b_last_conn,
lload_monitor_up_conn_entry, &arg );
e = mbe->entry_stub( &ms->mss_dn, &ms->mss_ndn, &bv_rdn,
oc_olmBalancerConnection, &bv_timestamp, &bv_timestamp );
connections_walk_last( &b->b_mutex, &b->b_bindconns, b->b_last_bindconn,
lload_monitor_up_conn_entry, &arg );
checked_unlock( &b->b_mutex );
cb = ch_calloc( sizeof(monitor_callback_t), 1 );
cb->mc_update = lload_monitor_conn_update;
cb->mc_private = c;
attr_merge_one( e, ad_olmConnectionType, &value, NULL );
attr_merge_one( e, ad_olmConnectionState, &value, NULL );
attr_merge_one( e, ad_olmPendingOps, &zero, NULL );
attr_merge_one( e, ad_olmReceivedOps, &zero, NULL );
attr_merge_one( e, ad_olmCompletedOps, &zero, NULL );
attr_merge_one( e, ad_olmFailedOps, &zero, NULL );
if ( mbe->register_entry( e, cb, NULL, 0 ) ) {
Debug( LDAP_DEBUG_ANY, "lload_monitor_conn_entry_create: "
"failed to register monitor entry for connid=%lu\n",
c->c_connid );
ch_free( cb );
entry_free( e );
return -1;
}
ber_dupbv( &c->c_monitor_dn, &e->e_nname );
entry_free( e );
return 0;
}
int
static int
lload_monitor_incoming_conn_init( BackendDB *be, monitor_subsys_t *ms )
{
monitor_extra_t *mbe;
......@@ -694,7 +736,6 @@ lload_monitor_incoming_conn_init( BackendDB *be, monitor_subsys_t *ms )
assert( be != NULL );
mbe = (monitor_extra_t *)be->bd_info->bi_extra;
ms->mss_create = lload_monitor_in_conn_create;
ms->mss_destroy = lload_monitor_subsystem_destroy;
dnNormalize( 0, NULL, NULL, &ms->mss_dn, &ms->mss_ndn, NULL );
......@@ -711,7 +752,7 @@ lload_monitor_incoming_conn_init( BackendDB *be, monitor_subsys_t *ms )
ber_dupbv( &ms->mss_dn, &e->e_name );
ber_dupbv( &ms->mss_ndn, &e->e_nname );
rc = mbe->register_entry( e, NULL, ms, MONITOR_F_VOLATILE_CH );
rc = mbe->register_entry( e, NULL, ms, 0 );
if ( rc != LDAP_SUCCESS ) {
Debug( LDAP_DEBUG_ANY, "lload_monitor_incoming_conn_init: "
......@@ -719,6 +760,9 @@ lload_monitor_incoming_conn_init( BackendDB *be, monitor_subsys_t *ms )
e->e_name.bv_val );
goto done;
}
lload_monitor_client_subsys = ms;
done:
entry_free( e );
......@@ -832,7 +876,7 @@ lload_monitor_backend_open( BackendDB *be, monitor_subsys_t *ms )
attr_merge_normalize_one( e, ad_olmCompletedOps, &value, NULL );
attr_merge_normalize_one( e, ad_olmFailedOps, &value, NULL );
rc = mbe->register_entry( e, cb, ms, MONITOR_F_VOLATILE_CH );
rc = mbe->register_entry( e, cb, ms, 0 );
if ( rc != LDAP_SUCCESS ) {
Debug( LDAP_DEBUG_ANY, "lload_monitor_backend_open: "
......@@ -841,7 +885,6 @@ lload_monitor_backend_open( BackendDB *be, monitor_subsys_t *ms )
goto done;
}
b->b_monitor = ms;
ms->mss_destroy = lload_monitor_backend_destroy;
done:
......@@ -867,9 +910,8 @@ lload_monitor_backend_init(
bk_mss->mss_rdn.bv_len, "cn=%s", b->b_name.bv_val );
bk_mss->mss_name = b->b_name.bv_val;
bk_mss->mss_flags = MONITOR_F_VOLATILE_CH;
bk_mss->mss_flags = MONITOR_F_NONE;
bk_mss->mss_open = lload_monitor_backend_open;
bk_mss->mss_create = lload_monitor_up_conn_create;
bk_mss->mss_destroy = lload_monitor_subsystem_destroy;
bk_mss->mss_update = NULL;
bk_mss->mss_private = b;
......@@ -878,8 +920,11 @@ lload_monitor_backend_init(
Debug( LDAP_DEBUG_ANY, "lload_monitor_backend_init: "
"failed to register backend %s\n",
bk_mss->mss_name );
ch_free( bk_mss );
return -1;
}
b->b_monitor = bk_mss;
return LDAP_SUCCESS;
}
......@@ -1102,7 +1147,7 @@ static struct monitor_subsys_t balancer_subsys[] = {
BER_BVNULL,
{ BER_BVC("Load Balancer incoming connections"),
BER_BVNULL },
MONITOR_F_VOLATILE_CH,
MONITOR_F_NONE,
lload_monitor_incoming_conn_init,
lload_monitor_subsystem_destroy, /* destroy */
NULL, /* update */
......
......@@ -163,7 +163,10 @@ LDAP_SLAPD_F (void) lload_libevent_destroy( void );
/*
* monitor.c
*/
LDAP_SLAPD_V (monitor_subsys_t *) lload_monitor_client_subsys;
LDAP_SLAPD_F (int) lload_monitor_open( void );
LDAP_SLAPD_F (int) lload_monitor_conn_entry_create( LloadConnection *c, monitor_subsys_t *ms );
LDAP_SLAPD_F (int) lload_monitor_conn_unlink( LloadConnection *c );
LDAP_SLAPD_F (int) lload_monitor_backend_init( BackendInfo *bi, monitor_subsys_t *ms, LloadBackend *b );
LDAP_SLAPD_F (int) lload_monitor_tier_init( BackendInfo *bi, LloadTier *tier );
#endif /* BALANCER_MODULE */
......
......@@ -966,6 +966,12 @@ upstream_init( ber_socket_t s, LloadBackend *b )
c->c_destroy = upstream_destroy;
c->c_unlink = upstream_unlink;
#ifdef BALANCER_MODULE
if ( b->b_monitor && lload_monitor_conn_entry_create( c, b->b_monitor ) ) {
goto fail;
}
#endif /* BALANCER_MODULE */
#ifdef HAVE_TLS
if ( c->c_is_tls == LLOAD_CLEARTEXT ) {
#endif /* HAVE_TLS */
......@@ -1122,6 +1128,17 @@ upstream_destroy( LloadConnection *c )
CONNECTION_LOCK(c);
assert( c->c_state == LLOAD_C_DYING );
#ifdef BALANCER_MODULE
/*
* Can't do this in upstream_unlink as that could be run from cn=monitor
* modify callback.
*/
if ( !BER_BVISNULL( &c->c_monitor_dn ) ) {
lload_monitor_conn_unlink( c );
}
#endif /* BALANCER_MODULE */
c->c_state = LLOAD_C_INVALID;
assert( c->c_ops == NULL );
......
......@@ -79,6 +79,7 @@ dn: cn=Connection 1,cn=backend,cn=first,cn=Backend Tiers,cn=Load Balancer,cn=B
ackends,cn=Monitor
objectClass: olmBalancerConnection
olmConnectionType: regular
olmConnectionState: ready
olmPendingOps: 0
olmReceivedOps: 0
olmCompletedOps: 0
......@@ -88,6 +89,7 @@ dn: cn=Connection 3,cn=backend,cn=first,cn=Backend Tiers,cn=Load Balancer,cn=B
ackends,cn=Monitor
objectClass: olmBalancerConnection
olmConnectionType: regular
olmConnectionState: ready
olmPendingOps: 0
olmReceivedOps: 0
olmCompletedOps: 0
......@@ -97,6 +99,7 @@ dn: cn=Connection 2,cn=backend,cn=first,cn=Backend Tiers,cn=Load Balancer,cn=B
ackends,cn=Monitor
objectClass: olmBalancerConnection
olmConnectionType: bind
olmConnectionState: ready
olmPendingOps: 0
olmReceivedOps: 0
olmCompletedOps: 0
......@@ -106,6 +109,7 @@ dn: cn=Connection 4,cn=backend,cn=first,cn=Backend Tiers,cn=Load Balancer,cn=B
ackends,cn=Monitor
objectClass: olmBalancerConnection
olmConnectionType: bind
olmConnectionState: ready
olmPendingOps: 0
olmReceivedOps: 0
olmCompletedOps: 0
......@@ -161,6 +165,7 @@ dn: cn=Connection 1,cn=backend,cn=first,cn=Backend Tiers,cn=Load Balancer,cn=B
ackends,cn=Monitor
objectClass: olmBalancerConnection
olmConnectionType: regular
olmConnectionState: ready
olmPendingOps: 0
olmReceivedOps: 0
olmCompletedOps: 0
......@@ -170,6 +175,7 @@ dn: cn=Connection 3,cn=backend,cn=first,cn=Backend Tiers,cn=Load Balancer,cn=B
ackends,cn=Monitor
objectClass: olmBalancerConnection
olmConnectionType: regular
olmConnectionState: ready
olmPendingOps: 0
olmReceivedOps: 0
olmCompletedOps: 0
......@@ -179,6 +185,7 @@ dn: cn=Connection 2,cn=backend,cn=first,cn=Backend Tiers,cn=Load Balancer,cn=B
ackends,cn=Monitor
objectClass: olmBalancerConnection
olmConnectionType: bind
olmConnectionState: ready
olmPendingOps: 0
olmReceivedOps: 1
olmCompletedOps: 1
......@@ -188,6 +195,7 @@ dn: cn=Connection 4,cn=backend,cn=first,cn=Backend Tiers,cn=Load Balancer,cn=B
ackends,cn=Monitor
objectClass: olmBalancerConnection
olmConnectionType: bind
olmConnectionState: ready
olmPendingOps: 0
olmReceivedOps: 1
olmCompletedOps: 1
......@@ -208,6 +216,7 @@ dn: cn=Connection 5,cn=server 2,cn=first,cn=Backend Tiers,cn=Load Balancer,cn=
Backends,cn=Monitor
objectClass: olmBalancerConnection
olmConnectionType: regular
olmConnectionState: ready
olmPendingOps: 0
olmReceivedOps: 1
olmCompletedOps: 1
......@@ -217,6 +226,7 @@ dn: cn=Connection 7,cn=server 2,cn=first,cn=Backend Tiers,cn=Load Balancer,cn=
Backends,cn=Monitor
objectClass: olmBalancerConnection
olmConnectionType: regular
olmConnectionState: ready
olmPendingOps: 0
olmReceivedOps: 1
olmCompletedOps: 1
......@@ -226,6 +236,7 @@ dn: cn=Connection 8,cn=server 2,cn=first,cn=Backend Tiers,cn=Load Balancer,cn=
Backends,cn=Monitor
objectClass: olmBalancerConnection
olmConnectionType: regular
olmConnectionState: ready
olmPendingOps: 0
olmReceivedOps: 0
olmCompletedOps: 0
......@@ -235,6 +246,7 @@ dn: cn=Connection 9,cn=server 2,cn=first,cn=Backend Tiers,cn=Load Balancer,cn=
Backends,cn=Monitor
objectClass: olmBalancerConnection
olmConnectionType: regular
olmConnectionState: ready
olmPendingOps: 0
olmReceivedOps: 0
olmCompletedOps: 0
......@@ -244,6 +256,7 @@ dn: cn=Connection 6,cn=server 2,cn=first,cn=Backend Tiers,cn=Load Balancer,cn=
Backends,cn=Monitor
objectClass: olmBalancerConnection
olmConnectionType: bind
olmConnectionState: ready
olmPendingOps: 0
olmReceivedOps: 0
olmCompletedOps: 0
......@@ -253,6 +266,7 @@ dn: cn=Connection 10,cn=server 2,cn=first,cn=Backend Tiers,cn=Load Balancer,cn
=Backends,cn=Monitor
objectClass: olmBalancerConnection