1 // SPDX-License-Identifier: GPL-2.0
2 #include <linux/ceph/ceph_debug.h>
4 #include <linux/module.h>
5 #include <linux/types.h>
6 #include <linux/slab.h>
7 #include <linux/random.h>
8 #include <linux/sched.h>
10 #include <linux/ceph/ceph_features.h>
11 #include <linux/ceph/mon_client.h>
12 #include <linux/ceph/libceph.h>
13 #include <linux/ceph/debugfs.h>
14 #include <linux/ceph/decode.h>
15 #include <linux/ceph/auth.h>
18 * Interact with Ceph monitor cluster. Handle requests for new map
19 * versions, and periodically resend as needed. Also implement
20 * statfs() and umount().
22 * A small cluster of Ceph "monitors" are responsible for managing critical
23 * cluster configuration and state information. An odd number (e.g., 3, 5)
24 * of cmon daemons use a modified version of the Paxos part-time parliament
25 * algorithm to manage the MDS map (mds cluster membership), OSD map, and
26 * list of clients who have mounted the file system.
28 * We maintain an open, active session with a monitor at all times in order to
29 * receive timely MDSMap updates. We periodically send a keepalive byte on the
30 * TCP socket to ensure we detect a failure. If the connection does break, we
31 * randomly hunt for a new monitor. Once the connection is reestablished, we
32 * resend any outstanding requests.
35 static const struct ceph_connection_operations mon_con_ops
;
37 static int __validate_auth(struct ceph_mon_client
*monc
);
39 static int decode_mon_info(void **p
, void *end
, bool msgr2
,
40 struct ceph_entity_addr
*addr
)
47 ret
= ceph_start_decoding(p
, end
, 1, "mon_info_t", &struct_v
,
52 mon_info_end
= *p
+ struct_len
;
53 ceph_decode_skip_string(p
, end
, e_inval
); /* skip mon name */
54 ret
= ceph_decode_entity_addrvec(p
, end
, msgr2
, addr
);
66 * Decode a monmap blob (e.g., during mount).
68 * Assume MonMap v3 (i.e. encoding with MONNAMES and MONENC).
70 static struct ceph_monmap
*ceph_monmap_decode(void **p
, void *end
, bool msgr2
)
72 struct ceph_monmap
*monmap
= NULL
;
73 struct ceph_fsid fsid
;
82 ceph_decode_32_safe(p
, end
, blob_len
, e_inval
);
83 ceph_decode_need(p
, end
, blob_len
, e_inval
);
85 ret
= ceph_start_decoding(p
, end
, 6, "monmap", &struct_v
, &struct_len
);
89 dout("%s struct_v %d\n", __func__
, struct_v
);
90 ceph_decode_copy_safe(p
, end
, &fsid
, sizeof(fsid
), e_inval
);
91 ceph_decode_32_safe(p
, end
, epoch
, e_inval
);
96 *p
+= sizeof(struct ceph_timespec
); /* skip last_changed */
97 *p
+= sizeof(struct ceph_timespec
); /* skip created */
99 ret
= ceph_start_decoding(p
, end
, 1, "mon_feature_t",
100 &feat_struct_v
, &feat_struct_len
);
104 *p
+= feat_struct_len
; /* skip persistent_features */
106 ret
= ceph_start_decoding(p
, end
, 1, "mon_feature_t",
107 &feat_struct_v
, &feat_struct_len
);
111 *p
+= feat_struct_len
; /* skip optional_features */
113 ceph_decode_32_safe(p
, end
, num_mon
, e_inval
);
115 dout("%s fsid %pU epoch %u num_mon %d\n", __func__
, &fsid
, epoch
,
117 if (num_mon
> CEPH_MAX_MON
)
120 monmap
= kmalloc(struct_size(monmap
, mon_inst
, num_mon
), GFP_NOIO
);
126 monmap
->epoch
= epoch
;
127 monmap
->num_mon
= num_mon
;
129 /* legacy_mon_addr map or mon_info map */
130 for (i
= 0; i
< num_mon
; i
++) {
131 struct ceph_entity_inst
*inst
= &monmap
->mon_inst
[i
];
133 ceph_decode_skip_string(p
, end
, e_inval
); /* skip mon name */
134 inst
->name
.type
= CEPH_ENTITY_TYPE_MON
;
135 inst
->name
.num
= cpu_to_le64(i
);
138 ret
= decode_mon_info(p
, end
, msgr2
, &inst
->addr
);
140 ret
= ceph_decode_entity_addr(p
, end
, &inst
->addr
);
144 dout("%s mon%d addr %s\n", __func__
, i
,
145 ceph_pr_addr(&inst
->addr
));
158 * return true if *addr is included in the monmap.
160 int ceph_monmap_contains(struct ceph_monmap
*m
, struct ceph_entity_addr
*addr
)
164 for (i
= 0; i
< m
->num_mon
; i
++) {
165 if (ceph_addr_equal_no_type(addr
, &m
->mon_inst
[i
].addr
))
173 * Send an auth request.
175 static void __send_prepared_auth_request(struct ceph_mon_client
*monc
, int len
)
177 monc
->pending_auth
= 1;
178 monc
->m_auth
->front
.iov_len
= len
;
179 monc
->m_auth
->hdr
.front_len
= cpu_to_le32(len
);
180 ceph_msg_revoke(monc
->m_auth
);
181 ceph_msg_get(monc
->m_auth
); /* keep our ref */
182 ceph_con_send(&monc
->con
, monc
->m_auth
);
186 * Close monitor session, if any.
188 static void __close_session(struct ceph_mon_client
*monc
)
190 dout("__close_session closing mon%d\n", monc
->cur_mon
);
191 ceph_msg_revoke(monc
->m_auth
);
192 ceph_msg_revoke_incoming(monc
->m_auth_reply
);
193 ceph_msg_revoke(monc
->m_subscribe
);
194 ceph_msg_revoke_incoming(monc
->m_subscribe_ack
);
195 ceph_con_close(&monc
->con
);
197 monc
->pending_auth
= 0;
198 ceph_auth_reset(monc
->auth
);
202 * Pick a new monitor at random and set cur_mon. If we are repicking
203 * (i.e. cur_mon is already set), be sure to pick a different one.
205 static void pick_new_mon(struct ceph_mon_client
*monc
)
207 int old_mon
= monc
->cur_mon
;
209 BUG_ON(monc
->monmap
->num_mon
< 1);
211 if (monc
->monmap
->num_mon
== 1) {
214 int max
= monc
->monmap
->num_mon
;
218 if (monc
->cur_mon
>= 0) {
219 if (monc
->cur_mon
< monc
->monmap
->num_mon
)
225 n
= get_random_u32_below(max
);
226 if (o
>= 0 && n
>= o
)
232 dout("%s mon%d -> mon%d out of %d mons\n", __func__
, old_mon
,
233 monc
->cur_mon
, monc
->monmap
->num_mon
);
237 * Open a session with a new monitor.
239 static void __open_session(struct ceph_mon_client
*monc
)
245 monc
->hunting
= true;
246 if (monc
->had_a_connection
) {
247 monc
->hunt_mult
*= CEPH_MONC_HUNT_BACKOFF
;
248 if (monc
->hunt_mult
> CEPH_MONC_HUNT_MAX_MULT
)
249 monc
->hunt_mult
= CEPH_MONC_HUNT_MAX_MULT
;
252 monc
->sub_renew_after
= jiffies
; /* i.e., expired */
253 monc
->sub_renew_sent
= 0;
255 dout("%s opening mon%d\n", __func__
, monc
->cur_mon
);
256 ceph_con_open(&monc
->con
, CEPH_ENTITY_TYPE_MON
, monc
->cur_mon
,
257 &monc
->monmap
->mon_inst
[monc
->cur_mon
].addr
);
260 * Queue a keepalive to ensure that in case of an early fault
261 * the messenger doesn't put us into STANDBY state and instead
262 * retries. This also ensures that our timestamp is valid by
263 * the time we finish hunting and delayed_work() checks it.
265 ceph_con_keepalive(&monc
->con
);
266 if (ceph_msgr2(monc
->client
)) {
267 monc
->pending_auth
= 1;
271 /* initiate authentication handshake */
272 ret
= ceph_auth_build_hello(monc
->auth
,
273 monc
->m_auth
->front
.iov_base
,
274 monc
->m_auth
->front_alloc_len
);
276 __send_prepared_auth_request(monc
, ret
);
279 static void reopen_session(struct ceph_mon_client
*monc
)
282 pr_info("mon%d %s session lost, hunting for new mon\n",
283 monc
->cur_mon
, ceph_pr_addr(&monc
->con
.peer_addr
));
285 __close_session(monc
);
286 __open_session(monc
);
289 void ceph_monc_reopen_session(struct ceph_mon_client
*monc
)
291 mutex_lock(&monc
->mutex
);
292 reopen_session(monc
);
293 mutex_unlock(&monc
->mutex
);
296 static void un_backoff(struct ceph_mon_client
*monc
)
298 monc
->hunt_mult
/= 2; /* reduce by 50% */
299 if (monc
->hunt_mult
< 1)
301 dout("%s hunt_mult now %d\n", __func__
, monc
->hunt_mult
);
305 * Reschedule delayed work timer.
307 static void __schedule_delayed(struct ceph_mon_client
*monc
)
312 delay
= CEPH_MONC_HUNT_INTERVAL
* monc
->hunt_mult
;
314 delay
= CEPH_MONC_PING_INTERVAL
;
316 dout("__schedule_delayed after %lu\n", delay
);
317 mod_delayed_work(system_wq
, &monc
->delayed_work
,
318 round_jiffies_relative(delay
));
321 const char *ceph_sub_str
[] = {
322 [CEPH_SUB_MONMAP
] = "monmap",
323 [CEPH_SUB_OSDMAP
] = "osdmap",
324 [CEPH_SUB_FSMAP
] = "fsmap.user",
325 [CEPH_SUB_MDSMAP
] = "mdsmap",
329 * Send subscribe request for one or more maps, according to
332 static void __send_subscribe(struct ceph_mon_client
*monc
)
334 struct ceph_msg
*msg
= monc
->m_subscribe
;
335 void *p
= msg
->front
.iov_base
;
336 void *const end
= p
+ msg
->front_alloc_len
;
340 dout("%s sent %lu\n", __func__
, monc
->sub_renew_sent
);
342 BUG_ON(monc
->cur_mon
< 0);
344 if (!monc
->sub_renew_sent
)
345 monc
->sub_renew_sent
= jiffies
| 1; /* never 0 */
347 msg
->hdr
.version
= cpu_to_le16(2);
349 for (i
= 0; i
< ARRAY_SIZE(monc
->subs
); i
++) {
350 if (monc
->subs
[i
].want
)
353 BUG_ON(num
< 1); /* monmap sub is always there */
354 ceph_encode_32(&p
, num
);
355 for (i
= 0; i
< ARRAY_SIZE(monc
->subs
); i
++) {
359 if (!monc
->subs
[i
].want
)
362 len
= sprintf(buf
, "%s", ceph_sub_str
[i
]);
363 if (i
== CEPH_SUB_MDSMAP
&&
364 monc
->fs_cluster_id
!= CEPH_FS_CLUSTER_ID_NONE
)
365 len
+= sprintf(buf
+ len
, ".%d", monc
->fs_cluster_id
);
367 dout("%s %s start %llu flags 0x%x\n", __func__
, buf
,
368 le64_to_cpu(monc
->subs
[i
].item
.start
),
369 monc
->subs
[i
].item
.flags
);
370 ceph_encode_string(&p
, end
, buf
, len
);
371 memcpy(p
, &monc
->subs
[i
].item
, sizeof(monc
->subs
[i
].item
));
372 p
+= sizeof(monc
->subs
[i
].item
);
376 msg
->front
.iov_len
= p
- msg
->front
.iov_base
;
377 msg
->hdr
.front_len
= cpu_to_le32(msg
->front
.iov_len
);
378 ceph_msg_revoke(msg
);
379 ceph_con_send(&monc
->con
, ceph_msg_get(msg
));
382 static void handle_subscribe_ack(struct ceph_mon_client
*monc
,
383 struct ceph_msg
*msg
)
385 unsigned int seconds
;
386 struct ceph_mon_subscribe_ack
*h
= msg
->front
.iov_base
;
388 if (msg
->front
.iov_len
< sizeof(*h
))
390 seconds
= le32_to_cpu(h
->duration
);
392 mutex_lock(&monc
->mutex
);
393 if (monc
->sub_renew_sent
) {
395 * This is only needed for legacy (infernalis or older)
396 * MONs -- see delayed_work().
398 monc
->sub_renew_after
= monc
->sub_renew_sent
+
399 (seconds
>> 1) * HZ
- 1;
400 dout("%s sent %lu duration %d renew after %lu\n", __func__
,
401 monc
->sub_renew_sent
, seconds
, monc
->sub_renew_after
);
402 monc
->sub_renew_sent
= 0;
404 dout("%s sent %lu renew after %lu, ignoring\n", __func__
,
405 monc
->sub_renew_sent
, monc
->sub_renew_after
);
407 mutex_unlock(&monc
->mutex
);
410 pr_err("got corrupt subscribe-ack msg\n");
415 * Register interest in a map
417 * @sub: one of CEPH_SUB_*
418 * @epoch: X for "every map since X", or 0 for "just the latest"
420 static bool __ceph_monc_want_map(struct ceph_mon_client
*monc
, int sub
,
421 u32 epoch
, bool continuous
)
423 __le64 start
= cpu_to_le64(epoch
);
424 u8 flags
= !continuous
? CEPH_SUBSCRIBE_ONETIME
: 0;
426 dout("%s %s epoch %u continuous %d\n", __func__
, ceph_sub_str
[sub
],
429 if (monc
->subs
[sub
].want
&&
430 monc
->subs
[sub
].item
.start
== start
&&
431 monc
->subs
[sub
].item
.flags
== flags
)
434 monc
->subs
[sub
].item
.start
= start
;
435 monc
->subs
[sub
].item
.flags
= flags
;
436 monc
->subs
[sub
].want
= true;
441 bool ceph_monc_want_map(struct ceph_mon_client
*monc
, int sub
, u32 epoch
,
446 mutex_lock(&monc
->mutex
);
447 need_request
= __ceph_monc_want_map(monc
, sub
, epoch
, continuous
);
448 mutex_unlock(&monc
->mutex
);
452 EXPORT_SYMBOL(ceph_monc_want_map
);
455 * Keep track of which maps we have
457 * @sub: one of CEPH_SUB_*
459 static void __ceph_monc_got_map(struct ceph_mon_client
*monc
, int sub
,
462 dout("%s %s epoch %u\n", __func__
, ceph_sub_str
[sub
], epoch
);
464 if (monc
->subs
[sub
].want
) {
465 if (monc
->subs
[sub
].item
.flags
& CEPH_SUBSCRIBE_ONETIME
)
466 monc
->subs
[sub
].want
= false;
468 monc
->subs
[sub
].item
.start
= cpu_to_le64(epoch
+ 1);
471 monc
->subs
[sub
].have
= epoch
;
474 void ceph_monc_got_map(struct ceph_mon_client
*monc
, int sub
, u32 epoch
)
476 mutex_lock(&monc
->mutex
);
477 __ceph_monc_got_map(monc
, sub
, epoch
);
478 mutex_unlock(&monc
->mutex
);
480 EXPORT_SYMBOL(ceph_monc_got_map
);
482 void ceph_monc_renew_subs(struct ceph_mon_client
*monc
)
484 mutex_lock(&monc
->mutex
);
485 __send_subscribe(monc
);
486 mutex_unlock(&monc
->mutex
);
488 EXPORT_SYMBOL(ceph_monc_renew_subs
);
491 * Wait for an osdmap with a given epoch.
493 * @epoch: epoch to wait for
494 * @timeout: in jiffies, 0 means "wait forever"
496 int ceph_monc_wait_osdmap(struct ceph_mon_client
*monc
, u32 epoch
,
497 unsigned long timeout
)
499 unsigned long started
= jiffies
;
502 mutex_lock(&monc
->mutex
);
503 while (monc
->subs
[CEPH_SUB_OSDMAP
].have
< epoch
) {
504 mutex_unlock(&monc
->mutex
);
506 if (timeout
&& time_after_eq(jiffies
, started
+ timeout
))
509 ret
= wait_event_interruptible_timeout(monc
->client
->auth_wq
,
510 monc
->subs
[CEPH_SUB_OSDMAP
].have
>= epoch
,
511 ceph_timeout_jiffies(timeout
));
515 mutex_lock(&monc
->mutex
);
518 mutex_unlock(&monc
->mutex
);
521 EXPORT_SYMBOL(ceph_monc_wait_osdmap
);
524 * Open a session with a random monitor. Request monmap and osdmap,
525 * which are waited upon in __ceph_open_session().
527 int ceph_monc_open_session(struct ceph_mon_client
*monc
)
529 mutex_lock(&monc
->mutex
);
530 __ceph_monc_want_map(monc
, CEPH_SUB_MONMAP
, 0, true);
531 __ceph_monc_want_map(monc
, CEPH_SUB_OSDMAP
, 0, false);
532 __open_session(monc
);
533 __schedule_delayed(monc
);
534 mutex_unlock(&monc
->mutex
);
537 EXPORT_SYMBOL(ceph_monc_open_session
);
539 static void ceph_monc_handle_map(struct ceph_mon_client
*monc
,
540 struct ceph_msg
*msg
)
542 struct ceph_client
*client
= monc
->client
;
543 struct ceph_monmap
*monmap
;
546 mutex_lock(&monc
->mutex
);
548 dout("handle_monmap\n");
549 p
= msg
->front
.iov_base
;
550 end
= p
+ msg
->front
.iov_len
;
552 monmap
= ceph_monmap_decode(&p
, end
, ceph_msgr2(client
));
553 if (IS_ERR(monmap
)) {
554 pr_err("problem decoding monmap, %d\n",
555 (int)PTR_ERR(monmap
));
560 if (ceph_check_fsid(client
, &monmap
->fsid
) < 0) {
566 monc
->monmap
= monmap
;
568 __ceph_monc_got_map(monc
, CEPH_SUB_MONMAP
, monc
->monmap
->epoch
);
569 client
->have_fsid
= true;
572 mutex_unlock(&monc
->mutex
);
573 wake_up_all(&client
->auth_wq
);
577 * generic requests (currently statfs, mon_get_version)
579 DEFINE_RB_FUNCS(generic_request
, struct ceph_mon_generic_request
, tid
, node
)
581 static void release_generic_request(struct kref
*kref
)
583 struct ceph_mon_generic_request
*req
=
584 container_of(kref
, struct ceph_mon_generic_request
, kref
);
586 dout("%s greq %p request %p reply %p\n", __func__
, req
, req
->request
,
588 WARN_ON(!RB_EMPTY_NODE(&req
->node
));
591 ceph_msg_put(req
->reply
);
593 ceph_msg_put(req
->request
);
598 static void put_generic_request(struct ceph_mon_generic_request
*req
)
601 kref_put(&req
->kref
, release_generic_request
);
604 static void get_generic_request(struct ceph_mon_generic_request
*req
)
606 kref_get(&req
->kref
);
609 static struct ceph_mon_generic_request
*
610 alloc_generic_request(struct ceph_mon_client
*monc
, gfp_t gfp
)
612 struct ceph_mon_generic_request
*req
;
614 req
= kzalloc(sizeof(*req
), gfp
);
619 kref_init(&req
->kref
);
620 RB_CLEAR_NODE(&req
->node
);
621 init_completion(&req
->completion
);
623 dout("%s greq %p\n", __func__
, req
);
627 static void register_generic_request(struct ceph_mon_generic_request
*req
)
629 struct ceph_mon_client
*monc
= req
->monc
;
633 get_generic_request(req
);
634 req
->tid
= ++monc
->last_tid
;
635 insert_generic_request(&monc
->generic_request_tree
, req
);
638 static void send_generic_request(struct ceph_mon_client
*monc
,
639 struct ceph_mon_generic_request
*req
)
643 dout("%s greq %p tid %llu\n", __func__
, req
, req
->tid
);
644 req
->request
->hdr
.tid
= cpu_to_le64(req
->tid
);
645 ceph_con_send(&monc
->con
, ceph_msg_get(req
->request
));
648 static void __finish_generic_request(struct ceph_mon_generic_request
*req
)
650 struct ceph_mon_client
*monc
= req
->monc
;
652 dout("%s greq %p tid %llu\n", __func__
, req
, req
->tid
);
653 erase_generic_request(&monc
->generic_request_tree
, req
);
655 ceph_msg_revoke(req
->request
);
656 ceph_msg_revoke_incoming(req
->reply
);
659 static void finish_generic_request(struct ceph_mon_generic_request
*req
)
661 __finish_generic_request(req
);
662 put_generic_request(req
);
665 static void complete_generic_request(struct ceph_mon_generic_request
*req
)
667 if (req
->complete_cb
)
668 req
->complete_cb(req
);
670 complete_all(&req
->completion
);
671 put_generic_request(req
);
674 static void cancel_generic_request(struct ceph_mon_generic_request
*req
)
676 struct ceph_mon_client
*monc
= req
->monc
;
677 struct ceph_mon_generic_request
*lookup_req
;
679 dout("%s greq %p tid %llu\n", __func__
, req
, req
->tid
);
681 mutex_lock(&monc
->mutex
);
682 lookup_req
= lookup_generic_request(&monc
->generic_request_tree
,
685 WARN_ON(lookup_req
!= req
);
686 finish_generic_request(req
);
689 mutex_unlock(&monc
->mutex
);
692 static int wait_generic_request(struct ceph_mon_generic_request
*req
)
696 dout("%s greq %p tid %llu\n", __func__
, req
, req
->tid
);
697 ret
= wait_for_completion_interruptible(&req
->completion
);
699 cancel_generic_request(req
);
701 ret
= req
->result
; /* completed */
706 static struct ceph_msg
*get_generic_reply(struct ceph_connection
*con
,
707 struct ceph_msg_header
*hdr
,
710 struct ceph_mon_client
*monc
= con
->private;
711 struct ceph_mon_generic_request
*req
;
712 u64 tid
= le64_to_cpu(hdr
->tid
);
715 mutex_lock(&monc
->mutex
);
716 req
= lookup_generic_request(&monc
->generic_request_tree
, tid
);
718 dout("get_generic_reply %lld dne\n", tid
);
722 dout("get_generic_reply %lld got %p\n", tid
, req
->reply
);
724 m
= ceph_msg_get(req
->reply
);
726 * we don't need to track the connection reading into
727 * this reply because we only have one open connection
731 mutex_unlock(&monc
->mutex
);
738 static void handle_statfs_reply(struct ceph_mon_client
*monc
,
739 struct ceph_msg
*msg
)
741 struct ceph_mon_generic_request
*req
;
742 struct ceph_mon_statfs_reply
*reply
= msg
->front
.iov_base
;
743 u64 tid
= le64_to_cpu(msg
->hdr
.tid
);
745 dout("%s msg %p tid %llu\n", __func__
, msg
, tid
);
747 if (msg
->front
.iov_len
!= sizeof(*reply
))
750 mutex_lock(&monc
->mutex
);
751 req
= lookup_generic_request(&monc
->generic_request_tree
, tid
);
753 mutex_unlock(&monc
->mutex
);
758 *req
->u
.st
= reply
->st
; /* struct */
759 __finish_generic_request(req
);
760 mutex_unlock(&monc
->mutex
);
762 complete_generic_request(req
);
766 pr_err("corrupt statfs reply, tid %llu\n", tid
);
771 * Do a synchronous statfs().
773 int ceph_monc_do_statfs(struct ceph_mon_client
*monc
, u64 data_pool
,
774 struct ceph_statfs
*buf
)
776 struct ceph_mon_generic_request
*req
;
777 struct ceph_mon_statfs
*h
;
780 req
= alloc_generic_request(monc
, GFP_NOFS
);
784 req
->request
= ceph_msg_new(CEPH_MSG_STATFS
, sizeof(*h
), GFP_NOFS
,
789 req
->reply
= ceph_msg_new(CEPH_MSG_STATFS_REPLY
, 64, GFP_NOFS
, true);
794 req
->request
->hdr
.version
= cpu_to_le16(2);
796 mutex_lock(&monc
->mutex
);
797 register_generic_request(req
);
798 /* fill out request */
799 h
= req
->request
->front
.iov_base
;
800 h
->monhdr
.have_version
= 0;
801 h
->monhdr
.session_mon
= cpu_to_le16(-1);
802 h
->monhdr
.session_mon_tid
= 0;
803 h
->fsid
= monc
->monmap
->fsid
;
804 h
->contains_data_pool
= (data_pool
!= CEPH_NOPOOL
);
805 h
->data_pool
= cpu_to_le64(data_pool
);
806 send_generic_request(monc
, req
);
807 mutex_unlock(&monc
->mutex
);
809 ret
= wait_generic_request(req
);
811 put_generic_request(req
);
814 EXPORT_SYMBOL(ceph_monc_do_statfs
);
816 static void handle_get_version_reply(struct ceph_mon_client
*monc
,
817 struct ceph_msg
*msg
)
819 struct ceph_mon_generic_request
*req
;
820 u64 tid
= le64_to_cpu(msg
->hdr
.tid
);
821 void *p
= msg
->front
.iov_base
;
822 void *end
= p
+ msg
->front_alloc_len
;
825 dout("%s msg %p tid %llu\n", __func__
, msg
, tid
);
827 ceph_decode_need(&p
, end
, 2*sizeof(u64
), bad
);
828 handle
= ceph_decode_64(&p
);
829 if (tid
!= 0 && tid
!= handle
)
832 mutex_lock(&monc
->mutex
);
833 req
= lookup_generic_request(&monc
->generic_request_tree
, handle
);
835 mutex_unlock(&monc
->mutex
);
840 req
->u
.newest
= ceph_decode_64(&p
);
841 __finish_generic_request(req
);
842 mutex_unlock(&monc
->mutex
);
844 complete_generic_request(req
);
848 pr_err("corrupt mon_get_version reply, tid %llu\n", tid
);
852 static struct ceph_mon_generic_request
*
853 __ceph_monc_get_version(struct ceph_mon_client
*monc
, const char *what
,
854 ceph_monc_callback_t cb
, u64 private_data
)
856 struct ceph_mon_generic_request
*req
;
858 req
= alloc_generic_request(monc
, GFP_NOIO
);
862 req
->request
= ceph_msg_new(CEPH_MSG_MON_GET_VERSION
,
863 sizeof(u64
) + sizeof(u32
) + strlen(what
),
868 req
->reply
= ceph_msg_new(CEPH_MSG_MON_GET_VERSION_REPLY
, 32, GFP_NOIO
,
873 req
->complete_cb
= cb
;
874 req
->private_data
= private_data
;
876 mutex_lock(&monc
->mutex
);
877 register_generic_request(req
);
879 void *p
= req
->request
->front
.iov_base
;
880 void *const end
= p
+ req
->request
->front_alloc_len
;
882 ceph_encode_64(&p
, req
->tid
); /* handle */
883 ceph_encode_string(&p
, end
, what
, strlen(what
));
886 send_generic_request(monc
, req
);
887 mutex_unlock(&monc
->mutex
);
892 put_generic_request(req
);
893 return ERR_PTR(-ENOMEM
);
897 * Send MMonGetVersion and wait for the reply.
899 * @what: one of "mdsmap", "osdmap" or "monmap"
901 int ceph_monc_get_version(struct ceph_mon_client
*monc
, const char *what
,
904 struct ceph_mon_generic_request
*req
;
907 req
= __ceph_monc_get_version(monc
, what
, NULL
, 0);
911 ret
= wait_generic_request(req
);
913 *newest
= req
->u
.newest
;
915 put_generic_request(req
);
918 EXPORT_SYMBOL(ceph_monc_get_version
);
921 * Send MMonGetVersion,
923 * @what: one of "mdsmap", "osdmap" or "monmap"
925 int ceph_monc_get_version_async(struct ceph_mon_client
*monc
, const char *what
,
926 ceph_monc_callback_t cb
, u64 private_data
)
928 struct ceph_mon_generic_request
*req
;
930 req
= __ceph_monc_get_version(monc
, what
, cb
, private_data
);
934 put_generic_request(req
);
937 EXPORT_SYMBOL(ceph_monc_get_version_async
);
939 static void handle_command_ack(struct ceph_mon_client
*monc
,
940 struct ceph_msg
*msg
)
942 struct ceph_mon_generic_request
*req
;
943 void *p
= msg
->front
.iov_base
;
944 void *const end
= p
+ msg
->front_alloc_len
;
945 u64 tid
= le64_to_cpu(msg
->hdr
.tid
);
947 dout("%s msg %p tid %llu\n", __func__
, msg
, tid
);
949 ceph_decode_need(&p
, end
, sizeof(struct ceph_mon_request_header
) +
951 p
+= sizeof(struct ceph_mon_request_header
);
953 mutex_lock(&monc
->mutex
);
954 req
= lookup_generic_request(&monc
->generic_request_tree
, tid
);
956 mutex_unlock(&monc
->mutex
);
960 req
->result
= ceph_decode_32(&p
);
961 __finish_generic_request(req
);
962 mutex_unlock(&monc
->mutex
);
964 complete_generic_request(req
);
968 pr_err("corrupt mon_command ack, tid %llu\n", tid
);
972 static __printf(2, 0)
973 int do_mon_command_vargs(struct ceph_mon_client
*monc
, const char *fmt
,
976 struct ceph_mon_generic_request
*req
;
977 struct ceph_mon_command
*h
;
981 req
= alloc_generic_request(monc
, GFP_NOIO
);
985 req
->request
= ceph_msg_new(CEPH_MSG_MON_COMMAND
, 256, GFP_NOIO
, true);
989 req
->reply
= ceph_msg_new(CEPH_MSG_MON_COMMAND_ACK
, 512, GFP_NOIO
,
994 mutex_lock(&monc
->mutex
);
995 register_generic_request(req
);
996 h
= req
->request
->front
.iov_base
;
997 h
->monhdr
.have_version
= 0;
998 h
->monhdr
.session_mon
= cpu_to_le16(-1);
999 h
->monhdr
.session_mon_tid
= 0;
1000 h
->fsid
= monc
->monmap
->fsid
;
1001 h
->num_strs
= cpu_to_le32(1);
1002 len
= vsprintf(h
->str
, fmt
, ap
);
1003 h
->str_len
= cpu_to_le32(len
);
1004 send_generic_request(monc
, req
);
1005 mutex_unlock(&monc
->mutex
);
1007 ret
= wait_generic_request(req
);
1009 put_generic_request(req
);
1013 static __printf(2, 3)
1014 int do_mon_command(struct ceph_mon_client
*monc
, const char *fmt
, ...)
1020 ret
= do_mon_command_vargs(monc
, fmt
, ap
);
1025 int ceph_monc_blocklist_add(struct ceph_mon_client
*monc
,
1026 struct ceph_entity_addr
*client_addr
)
1030 ret
= do_mon_command(monc
,
1031 "{ \"prefix\": \"osd blocklist\", \
1032 \"blocklistop\": \"add\", \
1033 \"addr\": \"%pISpc/%u\" }",
1034 &client_addr
->in_addr
,
1035 le32_to_cpu(client_addr
->nonce
));
1036 if (ret
== -EINVAL
) {
1038 * The monitor returns EINVAL on an unrecognized command.
1039 * Try the legacy command -- it is exactly the same except
1042 ret
= do_mon_command(monc
,
1043 "{ \"prefix\": \"osd blacklist\", \
1044 \"blacklistop\": \"add\", \
1045 \"addr\": \"%pISpc/%u\" }",
1046 &client_addr
->in_addr
,
1047 le32_to_cpu(client_addr
->nonce
));
1053 * Make sure we have the osdmap that includes the blocklist
1054 * entry. This is needed to ensure that the OSDs pick up the
1055 * new blocklist before processing any future requests from
1058 return ceph_wait_for_latest_osdmap(monc
->client
, 0);
1060 EXPORT_SYMBOL(ceph_monc_blocklist_add
);
1063 * Resend pending generic requests.
1065 static void __resend_generic_request(struct ceph_mon_client
*monc
)
1067 struct ceph_mon_generic_request
*req
;
1070 for (p
= rb_first(&monc
->generic_request_tree
); p
; p
= rb_next(p
)) {
1071 req
= rb_entry(p
, struct ceph_mon_generic_request
, node
);
1072 ceph_msg_revoke(req
->request
);
1073 ceph_msg_revoke_incoming(req
->reply
);
1074 ceph_con_send(&monc
->con
, ceph_msg_get(req
->request
));
1079 * Delayed work. If we haven't mounted yet, retry. Otherwise,
1080 * renew/retry subscription as needed (in case it is timing out, or we
1081 * got an ENOMEM). And keep the monitor connection alive.
1083 static void delayed_work(struct work_struct
*work
)
1085 struct ceph_mon_client
*monc
=
1086 container_of(work
, struct ceph_mon_client
, delayed_work
.work
);
1088 mutex_lock(&monc
->mutex
);
1089 dout("%s mon%d\n", __func__
, monc
->cur_mon
);
1090 if (monc
->cur_mon
< 0) {
1094 if (monc
->hunting
) {
1095 dout("%s continuing hunt\n", __func__
);
1096 reopen_session(monc
);
1098 int is_auth
= ceph_auth_is_authenticated(monc
->auth
);
1100 dout("%s is_authed %d\n", __func__
, is_auth
);
1101 if (ceph_con_keepalive_expired(&monc
->con
,
1102 CEPH_MONC_PING_TIMEOUT
)) {
1103 dout("monc keepalive timeout\n");
1105 reopen_session(monc
);
1108 if (!monc
->hunting
) {
1109 ceph_con_keepalive(&monc
->con
);
1110 __validate_auth(monc
);
1115 !(monc
->con
.peer_features
& CEPH_FEATURE_MON_STATEFUL_SUB
)) {
1116 unsigned long now
= jiffies
;
1118 dout("%s renew subs? now %lu renew after %lu\n",
1119 __func__
, now
, monc
->sub_renew_after
);
1120 if (time_after_eq(now
, monc
->sub_renew_after
))
1121 __send_subscribe(monc
);
1124 __schedule_delayed(monc
);
1127 mutex_unlock(&monc
->mutex
);
1131 * On startup, we build a temporary monmap populated with the IPs
1132 * provided by mount(2).
1134 static int build_initial_monmap(struct ceph_mon_client
*monc
)
1136 __le32 my_type
= ceph_msgr2(monc
->client
) ?
1137 CEPH_ENTITY_ADDR_TYPE_MSGR2
: CEPH_ENTITY_ADDR_TYPE_LEGACY
;
1138 struct ceph_options
*opt
= monc
->client
->options
;
1139 int num_mon
= opt
->num_mon
;
1142 /* build initial monmap */
1143 monc
->monmap
= kzalloc(struct_size(monc
->monmap
, mon_inst
, num_mon
),
1147 monc
->monmap
->num_mon
= num_mon
;
1149 for (i
= 0; i
< num_mon
; i
++) {
1150 struct ceph_entity_inst
*inst
= &monc
->monmap
->mon_inst
[i
];
1152 memcpy(&inst
->addr
.in_addr
, &opt
->mon_addr
[i
].in_addr
,
1153 sizeof(inst
->addr
.in_addr
));
1154 inst
->addr
.type
= my_type
;
1155 inst
->addr
.nonce
= 0;
1156 inst
->name
.type
= CEPH_ENTITY_TYPE_MON
;
1157 inst
->name
.num
= cpu_to_le64(i
);
1162 int ceph_monc_init(struct ceph_mon_client
*monc
, struct ceph_client
*cl
)
1167 memset(monc
, 0, sizeof(*monc
));
1169 mutex_init(&monc
->mutex
);
1171 err
= build_initial_monmap(monc
);
1176 /* authentication */
1177 monc
->auth
= ceph_auth_init(cl
->options
->name
, cl
->options
->key
,
1178 cl
->options
->con_modes
);
1179 if (IS_ERR(monc
->auth
)) {
1180 err
= PTR_ERR(monc
->auth
);
1183 monc
->auth
->want_keys
=
1184 CEPH_ENTITY_TYPE_AUTH
| CEPH_ENTITY_TYPE_MON
|
1185 CEPH_ENTITY_TYPE_OSD
| CEPH_ENTITY_TYPE_MDS
;
1189 monc
->m_subscribe_ack
= ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE_ACK
,
1190 sizeof(struct ceph_mon_subscribe_ack
),
1192 if (!monc
->m_subscribe_ack
)
1195 monc
->m_subscribe
= ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE
, 128,
1197 if (!monc
->m_subscribe
)
1198 goto out_subscribe_ack
;
1200 monc
->m_auth_reply
= ceph_msg_new(CEPH_MSG_AUTH_REPLY
, 4096,
1202 if (!monc
->m_auth_reply
)
1205 monc
->m_auth
= ceph_msg_new(CEPH_MSG_AUTH
, 4096, GFP_KERNEL
, true);
1206 monc
->pending_auth
= 0;
1208 goto out_auth_reply
;
1210 ceph_con_init(&monc
->con
, monc
, &mon_con_ops
,
1211 &monc
->client
->msgr
);
1214 monc
->had_a_connection
= false;
1215 monc
->hunt_mult
= 1;
1217 INIT_DELAYED_WORK(&monc
->delayed_work
, delayed_work
);
1218 monc
->generic_request_tree
= RB_ROOT
;
1221 monc
->fs_cluster_id
= CEPH_FS_CLUSTER_ID_NONE
;
1226 ceph_msg_put(monc
->m_auth_reply
);
1228 ceph_msg_put(monc
->m_subscribe
);
1230 ceph_msg_put(monc
->m_subscribe_ack
);
1232 ceph_auth_destroy(monc
->auth
);
1234 kfree(monc
->monmap
);
1238 EXPORT_SYMBOL(ceph_monc_init
);
1240 void ceph_monc_stop(struct ceph_mon_client
*monc
)
1244 mutex_lock(&monc
->mutex
);
1245 __close_session(monc
);
1246 monc
->hunting
= false;
1248 mutex_unlock(&monc
->mutex
);
1250 cancel_delayed_work_sync(&monc
->delayed_work
);
1253 * flush msgr queue before we destroy ourselves to ensure that:
1254 * - any work that references our embedded con is finished.
1255 * - any osd_client or other work that may reference an authorizer
1256 * finishes before we shut down the auth subsystem.
1260 ceph_auth_destroy(monc
->auth
);
1262 WARN_ON(!RB_EMPTY_ROOT(&monc
->generic_request_tree
));
1264 ceph_msg_put(monc
->m_auth
);
1265 ceph_msg_put(monc
->m_auth_reply
);
1266 ceph_msg_put(monc
->m_subscribe
);
1267 ceph_msg_put(monc
->m_subscribe_ack
);
1269 kfree(monc
->monmap
);
1271 EXPORT_SYMBOL(ceph_monc_stop
);
1273 static void finish_hunting(struct ceph_mon_client
*monc
)
1275 if (monc
->hunting
) {
1276 dout("%s found mon%d\n", __func__
, monc
->cur_mon
);
1277 monc
->hunting
= false;
1278 monc
->had_a_connection
= true;
1280 __schedule_delayed(monc
);
1284 static void finish_auth(struct ceph_mon_client
*monc
, int auth_err
,
1287 dout("%s auth_err %d was_authed %d\n", __func__
, auth_err
, was_authed
);
1288 WARN_ON(auth_err
> 0);
1290 monc
->pending_auth
= 0;
1292 monc
->client
->auth_err
= auth_err
;
1293 wake_up_all(&monc
->client
->auth_wq
);
1297 if (!was_authed
&& ceph_auth_is_authenticated(monc
->auth
)) {
1298 dout("%s authenticated, starting session global_id %llu\n",
1299 __func__
, monc
->auth
->global_id
);
1301 monc
->client
->msgr
.inst
.name
.type
= CEPH_ENTITY_TYPE_CLIENT
;
1302 monc
->client
->msgr
.inst
.name
.num
=
1303 cpu_to_le64(monc
->auth
->global_id
);
1305 __send_subscribe(monc
);
1306 __resend_generic_request(monc
);
1308 pr_info("mon%d %s session established\n", monc
->cur_mon
,
1309 ceph_pr_addr(&monc
->con
.peer_addr
));
1313 static void handle_auth_reply(struct ceph_mon_client
*monc
,
1314 struct ceph_msg
*msg
)
1319 mutex_lock(&monc
->mutex
);
1320 was_authed
= ceph_auth_is_authenticated(monc
->auth
);
1321 ret
= ceph_handle_auth_reply(monc
->auth
, msg
->front
.iov_base
,
1323 monc
->m_auth
->front
.iov_base
,
1324 monc
->m_auth
->front_alloc_len
);
1326 __send_prepared_auth_request(monc
, ret
);
1328 finish_auth(monc
, ret
, was_authed
);
1329 finish_hunting(monc
);
1331 mutex_unlock(&monc
->mutex
);
1334 static int __validate_auth(struct ceph_mon_client
*monc
)
1338 if (monc
->pending_auth
)
1341 ret
= ceph_build_auth(monc
->auth
, monc
->m_auth
->front
.iov_base
,
1342 monc
->m_auth
->front_alloc_len
);
1344 return ret
; /* either an error, or no need to authenticate */
1345 __send_prepared_auth_request(monc
, ret
);
1349 int ceph_monc_validate_auth(struct ceph_mon_client
*monc
)
1353 mutex_lock(&monc
->mutex
);
1354 ret
= __validate_auth(monc
);
1355 mutex_unlock(&monc
->mutex
);
1358 EXPORT_SYMBOL(ceph_monc_validate_auth
);
1360 static int mon_get_auth_request(struct ceph_connection
*con
,
1361 void *buf
, int *buf_len
,
1362 void **authorizer
, int *authorizer_len
)
1364 struct ceph_mon_client
*monc
= con
->private;
1367 mutex_lock(&monc
->mutex
);
1368 ret
= ceph_auth_get_request(monc
->auth
, buf
, *buf_len
);
1369 mutex_unlock(&monc
->mutex
);
1375 *authorizer_len
= 0;
1379 static int mon_handle_auth_reply_more(struct ceph_connection
*con
,
1380 void *reply
, int reply_len
,
1381 void *buf
, int *buf_len
,
1382 void **authorizer
, int *authorizer_len
)
1384 struct ceph_mon_client
*monc
= con
->private;
1387 mutex_lock(&monc
->mutex
);
1388 ret
= ceph_auth_handle_reply_more(monc
->auth
, reply
, reply_len
,
1390 mutex_unlock(&monc
->mutex
);
1396 *authorizer_len
= 0;
1400 static int mon_handle_auth_done(struct ceph_connection
*con
,
1401 u64 global_id
, void *reply
, int reply_len
,
1402 u8
*session_key
, int *session_key_len
,
1403 u8
*con_secret
, int *con_secret_len
)
1405 struct ceph_mon_client
*monc
= con
->private;
1409 mutex_lock(&monc
->mutex
);
1410 WARN_ON(!monc
->hunting
);
1411 was_authed
= ceph_auth_is_authenticated(monc
->auth
);
1412 ret
= ceph_auth_handle_reply_done(monc
->auth
, global_id
,
1414 session_key
, session_key_len
,
1415 con_secret
, con_secret_len
);
1416 finish_auth(monc
, ret
, was_authed
);
1418 finish_hunting(monc
);
1419 mutex_unlock(&monc
->mutex
);
1423 static int mon_handle_auth_bad_method(struct ceph_connection
*con
,
1424 int used_proto
, int result
,
1425 const int *allowed_protos
, int proto_cnt
,
1426 const int *allowed_modes
, int mode_cnt
)
1428 struct ceph_mon_client
*monc
= con
->private;
1431 mutex_lock(&monc
->mutex
);
1432 WARN_ON(!monc
->hunting
);
1433 was_authed
= ceph_auth_is_authenticated(monc
->auth
);
1434 ceph_auth_handle_bad_method(monc
->auth
, used_proto
, result
,
1435 allowed_protos
, proto_cnt
,
1436 allowed_modes
, mode_cnt
);
1437 finish_auth(monc
, -EACCES
, was_authed
);
1438 mutex_unlock(&monc
->mutex
);
1443 * handle incoming message
1445 static void mon_dispatch(struct ceph_connection
*con
, struct ceph_msg
*msg
)
1447 struct ceph_mon_client
*monc
= con
->private;
1448 int type
= le16_to_cpu(msg
->hdr
.type
);
1451 case CEPH_MSG_AUTH_REPLY
:
1452 handle_auth_reply(monc
, msg
);
1455 case CEPH_MSG_MON_SUBSCRIBE_ACK
:
1456 handle_subscribe_ack(monc
, msg
);
1459 case CEPH_MSG_STATFS_REPLY
:
1460 handle_statfs_reply(monc
, msg
);
1463 case CEPH_MSG_MON_GET_VERSION_REPLY
:
1464 handle_get_version_reply(monc
, msg
);
1467 case CEPH_MSG_MON_COMMAND_ACK
:
1468 handle_command_ack(monc
, msg
);
1471 case CEPH_MSG_MON_MAP
:
1472 ceph_monc_handle_map(monc
, msg
);
1475 case CEPH_MSG_OSD_MAP
:
1476 ceph_osdc_handle_map(&monc
->client
->osdc
, msg
);
1480 /* can the chained handler handle it? */
1481 if (monc
->client
->extra_mon_dispatch
&&
1482 monc
->client
->extra_mon_dispatch(monc
->client
, msg
) == 0)
1485 pr_err("received unknown message type %d %s\n", type
,
1486 ceph_msg_type_name(type
));
1492 * Allocate memory for incoming message
1494 static struct ceph_msg
*mon_alloc_msg(struct ceph_connection
*con
,
1495 struct ceph_msg_header
*hdr
,
1498 struct ceph_mon_client
*monc
= con
->private;
1499 int type
= le16_to_cpu(hdr
->type
);
1500 int front_len
= le32_to_cpu(hdr
->front_len
);
1501 struct ceph_msg
*m
= NULL
;
1506 case CEPH_MSG_MON_SUBSCRIBE_ACK
:
1507 m
= ceph_msg_get(monc
->m_subscribe_ack
);
1509 case CEPH_MSG_STATFS_REPLY
:
1510 case CEPH_MSG_MON_COMMAND_ACK
:
1511 return get_generic_reply(con
, hdr
, skip
);
1512 case CEPH_MSG_AUTH_REPLY
:
1513 m
= ceph_msg_get(monc
->m_auth_reply
);
1515 case CEPH_MSG_MON_GET_VERSION_REPLY
:
1516 if (le64_to_cpu(hdr
->tid
) != 0)
1517 return get_generic_reply(con
, hdr
, skip
);
1520 * Older OSDs don't set reply tid even if the original
1521 * request had a non-zero tid. Work around this weirdness
1522 * by allocating a new message.
1525 case CEPH_MSG_MON_MAP
:
1526 case CEPH_MSG_MDS_MAP
:
1527 case CEPH_MSG_OSD_MAP
:
1528 case CEPH_MSG_FS_MAP_USER
:
1529 m
= ceph_msg_new(type
, front_len
, GFP_NOFS
, false);
1531 return NULL
; /* ENOMEM--return skip == 0 */
1536 pr_info("alloc_msg unknown type %d\n", type
);
1538 } else if (front_len
> m
->front_alloc_len
) {
1539 pr_warn("mon_alloc_msg front %d > prealloc %d (%u#%llu)\n",
1540 front_len
, m
->front_alloc_len
,
1541 (unsigned int)con
->peer_name
.type
,
1542 le64_to_cpu(con
->peer_name
.num
));
1544 m
= ceph_msg_new(type
, front_len
, GFP_NOFS
, false);
1551 * If the monitor connection resets, pick a new monitor and resubmit
1552 * any pending requests.
1554 static void mon_fault(struct ceph_connection
*con
)
1556 struct ceph_mon_client
*monc
= con
->private;
1558 mutex_lock(&monc
->mutex
);
1559 dout("%s mon%d\n", __func__
, monc
->cur_mon
);
1560 if (monc
->cur_mon
>= 0) {
1561 if (!monc
->hunting
) {
1562 dout("%s hunting for new mon\n", __func__
);
1563 reopen_session(monc
);
1564 __schedule_delayed(monc
);
1566 dout("%s already hunting\n", __func__
);
1569 mutex_unlock(&monc
->mutex
);
1573 * We can ignore refcounting on the connection struct, as all references
1574 * will come from the messenger workqueue, which is drained prior to
1575 * mon_client destruction.
1577 static struct ceph_connection
*mon_get_con(struct ceph_connection
*con
)
1582 static void mon_put_con(struct ceph_connection
*con
)
1586 static const struct ceph_connection_operations mon_con_ops
= {
1589 .alloc_msg
= mon_alloc_msg
,
1590 .dispatch
= mon_dispatch
,
1592 .get_auth_request
= mon_get_auth_request
,
1593 .handle_auth_reply_more
= mon_handle_auth_reply_more
,
1594 .handle_auth_done
= mon_handle_auth_done
,
1595 .handle_auth_bad_method
= mon_handle_auth_bad_method
,