Skip to content

Commit 5ad254f

Browse files
chu11chu11
committed
job-info: route update-lookup to lookup target
Problem: Now that the job-info.lookup target takes the DECODE and CURRENT flags, it can behave just like the job-info.update-lookup target. Deprecate the job-info.update-lookup RPC target by simply having it call the job-info.lookup code path with the DECODE and CURRENT flags.
1 parent d229f2b commit 5ad254f

File tree

8 files changed

+287
-205
lines changed

8 files changed

+287
-205
lines changed

‎src/modules/job-info/job-info.c‎

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ static void stats_cb (flux_t *h, flux_msg_handler_t *mh,
3838
int lookups = zlist_size (ctx->lookups);
3939
int watchers = zlist_size (ctx->watchers);
4040
int guest_watchers = zlist_size (ctx->guest_watchers);
41-
int update_lookups = zlist_size (ctx->update_lookups);
41+
int update_lookups = 0; /* no longer supported */
4242
int update_watchers = update_watch_count (ctx);
4343
if (flux_respond_pack (h, msg, "{s:i s:i s:i s:i s:i}",
4444
"lookups", lookups,
@@ -117,10 +117,6 @@ static void info_ctx_destroy (struct info_ctx *ctx)
117117
guest_watch_cleanup (ctx);
118118
zlist_destroy (&ctx->guest_watchers);
119119
}
120-
if (ctx->update_lookups) {
121-
update_lookup_cleanup (ctx);
122-
zlist_destroy (&ctx->update_lookups);
123-
}
124120
if (ctx->update_watchers) {
125121
update_watch_cleanup (ctx);
126122
zlist_destroy (&ctx->update_watchers);
@@ -149,8 +145,6 @@ static struct info_ctx *info_ctx_create (flux_t *h)
149145
goto error;
150146
if (!(ctx->guest_watchers = zlist_new ()))
151147
goto error;
152-
if (!(ctx->update_lookups = zlist_new ()))
153-
goto error;
154148
if (!(ctx->update_watchers = zlist_new ()))
155149
goto error;
156150
/* no destructor for index_uw, destruction handled on

‎src/modules/job-info/job-info.h‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@ struct info_ctx {
2626
zlist_t *lookups;
2727
zlist_t *watchers;
2828
zlist_t *guest_watchers;
29-
zlist_t *update_lookups;
3029
zlist_t *update_watchers;
3130
zhashx_t *index_uw; /* update_watchers lookup */
3231
};

‎src/modules/job-info/lookup.c‎

Lines changed: 186 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -20,12 +20,13 @@
2020
#include "src/common/libczmqcontainers/czmq_containers.h"
2121
#include "src/common/libeventlog/eventlog.h"
2222
#include "src/common/libjob/idf58.h"
23-
#include "src/common/libutil/jpath.h"
2423
#include "ccan/str/str.h"
2524

2625
#include "job-info.h"
2726
#include "lookup.h"
27+
#include "update.h"
2828
#include "allow.h"
29+
#include "util.h"
2930

3031
struct lookup_ctx {
3132
struct info_ctx *ctx;
@@ -154,29 +155,6 @@ static int lookup_keys (struct lookup_ctx *l)
154155
return -1;
155156
}
156157

157-
static void apply_updates_R (struct lookup_ctx *l,
158-
const char *key,
159-
json_t *update_object,
160-
json_t *context)
161-
{
162-
const char *context_key;
163-
json_t *value;
164-
165-
json_object_foreach (context, context_key, value) {
166-
/* RFC 21 resource-update event only allows update
167-
* to:
168-
* - expiration
169-
*/
170-
if (streq (context_key, "expiration"))
171-
if (jpath_set (update_object,
172-
"execution.expiration",
173-
value) < 0)
174-
flux_log (l->ctx->h, LOG_INFO,
175-
"%s: failed to update job %s %s",
176-
__FUNCTION__, idf58 (l->id), key);
177-
}
178-
}
179-
180158
static int lookup_current (struct lookup_ctx *l,
181159
flux_future_t *fall,
182160
const char *key,
@@ -226,7 +204,7 @@ static int lookup_current (struct lookup_ctx *l,
226204
goto error;
227205
if (streq (name, update_event_name)) {
228206
if (streq (key, "R"))
229-
apply_updates_R (l, key, value_object, context);
207+
apply_updates_R (l->ctx->h, l->id, key, value_object, context);
230208
}
231209
}
232210

@@ -413,11 +391,148 @@ static int check_to_lookup_eventlog (struct lookup_ctx *l)
413391
return 0;
414392
}
415393

394+
static json_t *get_json_string (json_t *o)
395+
{
396+
char *s = json_dumps (o, JSON_ENCODE_ANY);
397+
json_t *tmp = NULL;
398+
/* We assume json is internally valid, thus this is an ENOMEM error */
399+
if (!s) {
400+
errno = ENOMEM;
401+
goto cleanup;
402+
}
403+
if (!(tmp = json_string (s))) {
404+
errno = ENOMEM;
405+
goto cleanup;
406+
}
407+
cleanup:
408+
free (s);
409+
return tmp;
410+
}
411+
412+
/* returns -1 on error, 1 on cached response returned, 0 on no cache */
413+
static int lookup_cached (struct lookup_ctx *l)
414+
{
415+
json_t *current_object = NULL;
416+
json_t *key;
417+
const char *key_str;
418+
int ret, rv = -1;
419+
420+
/* Special optimization, looking for a single updated value that
421+
* could be cached via an update-watch
422+
*
423+
* - Caller must want current / updated value
424+
* - This lookup is already allowed (i.e. if we have to do a
425+
* "allow" KVS lookup, there is little benefit to returning the
426+
* cached value).
427+
* - The caller only wants one key (i.e. if we have to do lookup
428+
* on another value anyways, there is little benefit to
429+
* returning the cached value).
430+
*/
431+
432+
if (!(l->flags & FLUX_JOB_LOOKUP_CURRENT)
433+
|| !l->allow
434+
|| json_array_size (l->keys) != 1)
435+
return 0;
436+
437+
key = json_array_get (l->keys, 0);
438+
if (!key) {
439+
errno = EINVAL;
440+
goto cleanup;
441+
}
442+
443+
key_str = json_string_value (key);
444+
445+
if (!streq (key_str, "R"))
446+
return 0;
447+
448+
if ((ret = update_watch_get_cached (l->ctx,
449+
l->id,
450+
key_str,
451+
&current_object)) < 0)
452+
goto cleanup;
453+
454+
if (ret) {
455+
if (l->flags & FLUX_JOB_LOOKUP_JSON_DECODE) {
456+
if (flux_respond_pack (l->ctx->h, l->msg, "{s:I s:O}",
457+
"id", l->id,
458+
key_str, current_object) < 0) {
459+
flux_log_error (l->ctx->h, "%s: flux_respond", __FUNCTION__);
460+
goto cleanup;
461+
}
462+
rv = 1;
463+
goto cleanup;
464+
}
465+
else {
466+
json_t *o = get_json_string (current_object);
467+
if (!o) {
468+
errno = ENOMEM;
469+
goto cleanup;
470+
}
471+
if (flux_respond_pack (l->ctx->h, l->msg, "{s:I s:O}",
472+
"id", l->id,
473+
key_str, o) < 0) {
474+
json_decref (o);
475+
flux_log_error (l->ctx->h, "%s: flux_respond", __FUNCTION__);
476+
goto cleanup;
477+
}
478+
rv = 1;
479+
json_decref (o);
480+
goto cleanup;
481+
}
482+
}
483+
484+
rv = 0;
485+
cleanup:
486+
json_decref (current_object);
487+
return rv;
488+
}
489+
490+
static int lookup (flux_t *h,
491+
const flux_msg_t *msg,
492+
struct info_ctx *ctx,
493+
flux_jobid_t id,
494+
json_t *keys,
495+
int flags)
496+
{
497+
struct lookup_ctx *l = NULL;
498+
int ret;
499+
500+
if (!(l = lookup_ctx_create (ctx, msg, id, keys, flags)))
501+
goto error;
502+
503+
if (check_allow (l) < 0)
504+
goto error;
505+
506+
if ((ret = lookup_cached (l)) < 0)
507+
goto error;
508+
509+
if (ret) {
510+
lookup_ctx_destroy (l);
511+
return 0;
512+
}
513+
514+
if (check_to_lookup_eventlog (l) < 0)
515+
goto error;
516+
517+
if (lookup_keys (l) < 0)
518+
goto error;
519+
520+
if (zlist_append (ctx->lookups, l) < 0) {
521+
flux_log_error (h, "%s: zlist_append", __FUNCTION__);
522+
goto error;
523+
}
524+
zlist_freefn (ctx->lookups, l, lookup_ctx_destroy, true);
525+
return 0;
526+
527+
error:
528+
lookup_ctx_destroy (l);
529+
return -1;
530+
}
531+
416532
void lookup_cb (flux_t *h, flux_msg_handler_t *mh,
417533
const flux_msg_t *msg, void *arg)
418534
{
419535
struct info_ctx *ctx = arg;
420-
struct lookup_ctx *l = NULL;
421536
size_t index;
422537
json_t *key;
423538
json_t *keys;
@@ -453,30 +568,65 @@ void lookup_cb (flux_t *h, flux_msg_handler_t *mh,
453568
}
454569
}
455570

456-
if (!(l = lookup_ctx_create (ctx, msg, id, keys, flags)))
571+
if (lookup (h, msg, ctx, id, keys, flags) < 0)
457572
goto error;
458573

459-
if (check_allow (l) < 0)
460-
goto error;
574+
return;
461575

462-
if (check_to_lookup_eventlog (l) < 0)
463-
goto error;
576+
error:
577+
if (flux_respond_error (h, msg, errno, errmsg) < 0)
578+
flux_log_error (h, "%s: flux_respond_error", __FUNCTION__);
579+
}
464580

465-
if (lookup_keys (l) < 0)
581+
/* legacy rpc target */
582+
void update_lookup_cb (flux_t *h, flux_msg_handler_t *mh,
583+
const flux_msg_t *msg, void *arg)
584+
{
585+
struct info_ctx *ctx = arg;
586+
flux_jobid_t id;
587+
const char *key = NULL;
588+
json_t *keys = NULL;
589+
int flags;
590+
int valid_flags = 0;
591+
const char *errmsg = NULL;
592+
593+
if (flux_request_unpack (msg, NULL, "{s:I s:s s:i}",
594+
"id", &id,
595+
"key", &key,
596+
"flags", &flags) < 0) {
597+
flux_log_error (h, "%s: flux_request_unpack", __FUNCTION__);
466598
goto error;
599+
}
600+
if ((flags & ~valid_flags)) {
601+
errno = EPROTO;
602+
errmsg = "update-lookup request rejected with invalid flag";
603+
goto error;
604+
}
605+
if (!streq (key, "R")) {
606+
errno = EINVAL;
607+
errmsg = "update-lookup unsupported key specified";
608+
goto error;
609+
}
467610

468-
if (zlist_append (ctx->lookups, l) < 0) {
469-
flux_log_error (h, "%s: zlist_append", __FUNCTION__);
611+
if (!(keys = json_pack ("[s]", key))) {
612+
errno = ENOMEM;
470613
goto error;
471614
}
472-
zlist_freefn (ctx->lookups, l, lookup_ctx_destroy, true);
615+
616+
if (lookup (h,
617+
msg,
618+
ctx,
619+
id,
620+
keys,
621+
FLUX_JOB_LOOKUP_JSON_DECODE | FLUX_JOB_LOOKUP_CURRENT) < 0)
622+
goto error;
473623

474624
return;
475625

476626
error:
477627
if (flux_respond_error (h, msg, errno, errmsg) < 0)
478628
flux_log_error (h, "%s: flux_respond_error", __FUNCTION__);
479-
lookup_ctx_destroy (l);
629+
json_decref (keys);
480630
}
481631

482632
/*

‎src/modules/job-info/lookup.h‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,10 @@
1616
void lookup_cb (flux_t *h, flux_msg_handler_t *mh,
1717
const flux_msg_t *msg, void *arg);
1818

19+
/* legacy rpc target */
20+
void update_lookup_cb (flux_t *h, flux_msg_handler_t *mh,
21+
const flux_msg_t *msg, void *arg);
22+
1923
#endif /* ! _FLUX_JOB_INFO_LOOKUP_H */
2024

2125
/*

0 commit comments

Comments
 (0)