rabbitmq-server/moduleindex.yaml

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

1329 lines
29 KiB
YAML
Raw Permalink Normal View History

accept:
- accept_encoding_header
- accept_header
- accept_neg
- accept_parser
amqp_client:
- amqp_auth_mechanisms
- amqp_channel
- amqp_channel_sup
- amqp_channel_sup_sup
- amqp_channels_manager
- amqp_client
- amqp_connection
- amqp_connection_sup
- amqp_connection_type_sup
- amqp_direct_connection
- amqp_direct_consumer
- amqp_gen_connection
- amqp_gen_consumer
- amqp_main_reader
- amqp_network_connection
- amqp_rpc_client
- amqp_rpc_server
- amqp_selective_consumer
- amqp_ssl
- amqp_sup
- amqp_uri
- amqp_util
- rabbit_routing_util
- uri_parser
amqp10_client:
- amqp10_client
- amqp10_client_app
- amqp10_client_connection
- amqp10_client_connection_sup
- amqp10_client_frame_reader
- amqp10_client_session
- amqp10_client_sessions_sup
- amqp10_client_sup
- amqp10_client_types
- amqp10_msg
amqp10_common:
- amqp10_binary_generator
- amqp10_binary_parser
- amqp10_framing
- amqp10_framing0
- amqp10_util
- serial_number
aten:
- aten
- aten_app
- aten_detect
- aten_detector
- aten_emitter
- aten_sink
- aten_sup
base64url:
- base64url
cowboy:
- cowboy
- cowboy_app
- cowboy_bstr
- cowboy_children
- cowboy_clear
- cowboy_clock
- cowboy_compress_h
- cowboy_constraints
- cowboy_decompress_h
- cowboy_handler
- cowboy_http
- cowboy_http2
- cowboy_loop
- cowboy_metrics_h
- cowboy_middleware
- cowboy_req
- cowboy_rest
- cowboy_router
- cowboy_static
- cowboy_stream
- cowboy_stream_h
- cowboy_sub_protocol
- cowboy_sup
- cowboy_tls
- cowboy_tracer_h
- cowboy_websocket
cowlib:
- cow_base64url
- cow_cookie
- cow_date
- cow_hpack
- cow_http
- cow_http2
- cow_http2_machine
- cow_http_hd
- cow_http_struct_hd
- cow_http_te
- cow_iolists
- cow_link
- cow_mimetypes
- cow_multipart
- cow_qs
- cow_spdy
- cow_sse
- cow_uri
- cow_uri_template
- cow_ws
credentials_obfuscation:
- credentials_obfuscation
- credentials_obfuscation_app
- credentials_obfuscation_pbe
- credentials_obfuscation_sup
- credentials_obfuscation_svc
ct_helper:
- ct_helper
- ct_helper_error_h
cuttlefish:
- conf_parse
- cuttlefish
- cuttlefish_advanced
- cuttlefish_bytesize
- cuttlefish_conf
- cuttlefish_datatypes
- cuttlefish_duration
- cuttlefish_duration_parse
- cuttlefish_effective
- cuttlefish_enum
- cuttlefish_error
- cuttlefish_escript
- cuttlefish_flag
- cuttlefish_generator
- cuttlefish_mapping
- cuttlefish_rebar_plugin
- cuttlefish_schema
- cuttlefish_translation
- cuttlefish_unit
- cuttlefish_util
- cuttlefish_validator
- cuttlefish_variable
- cuttlefish_vmargs
eetcd:
- auth_pb
- eetcd
- eetcd_app
- eetcd_auth
- eetcd_auth_gen
- eetcd_cluster
- eetcd_cluster_gen
- eetcd_compare
- eetcd_conn
- eetcd_conn_sup
- eetcd_data_coercion
- eetcd_election
- eetcd_election_gen
- eetcd_grpc
- eetcd_health_gen
- eetcd_kv
- eetcd_kv_gen
- eetcd_lease
- eetcd_lease_gen
- eetcd_lease_sup
- eetcd_lock
- eetcd_lock_gen
- eetcd_maintenance
- eetcd_maintenance_gen
- eetcd_op
- eetcd_stream
- eetcd_sup
- eetcd_watch
- eetcd_watch_gen
- gogo_pb
- health_pb
- kv_pb
- router_pb
emqtt:
- emqtt
- emqtt_cli
- emqtt_frame
- emqtt_inflight
- emqtt_props
- emqtt_quic
- emqtt_quic_connection
- emqtt_quic_stream
- emqtt_secret
- emqtt_sock
- emqtt_ws
enough:
- enough
eunit_formatters:
- binomial_heap
- eunit_progress
gen_batch_server:
- gen_batch_server
getopt:
- getopt
gun:
- gun
- gun_app
- gun_content_handler
- gun_data_h
- gun_http
- gun_http2
- gun_sse_h
- gun_sup
- gun_tcp
- gun_tls
- gun_ws
- gun_ws_h
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
horus:
- horus
- horus_cover
- horus_utils
host_triple:
- host_triple
inet_tcp_proxy_dist:
- inet_tcp_proxy_dist
- inet_tcp_proxy_dist_app
- inet_tcp_proxy_dist_conn_sup
- inet_tcp_proxy_dist_controller
- inet_tcp_proxy_dist_sup
jose:
- jose
- jose_app
- jose_base64
- jose_base64url
- jose_block_encryptor
- jose_chacha20_poly1305
- jose_chacha20_poly1305_crypto
- jose_chacha20_poly1305_libsodium
- jose_chacha20_poly1305_unsupported
- jose_crypto_compat
- jose_curve25519
- jose_curve25519_crypto
- jose_curve25519_fallback
- jose_curve25519_libdecaf
- jose_curve25519_libsodium
- jose_curve25519_unsupported
- jose_curve448
- jose_curve448_crypto
- jose_curve448_fallback
- jose_curve448_libdecaf
- jose_curve448_unsupported
- jose_json
- jose_json_jason
- jose_json_jiffy
- jose_json_jsone
- jose_json_jsx
- jose_json_ojson
- jose_json_poison
- jose_json_poison_compat_encoder
- jose_json_poison_lexical_encoder
- jose_json_thoas
- jose_json_unsupported
- jose_jwa
- jose_jwa_aes
- jose_jwa_aes_kw
- jose_jwa_base64url
- jose_jwa_bench
- jose_jwa_chacha20
- jose_jwa_chacha20_poly1305
- jose_jwa_concat_kdf
- jose_jwa_curve25519
- jose_jwa_curve448
- jose_jwa_ed25519
- jose_jwa_ed448
- jose_jwa_hchacha20
- jose_jwa_math
- jose_jwa_pkcs1
- jose_jwa_pkcs5
- jose_jwa_pkcs7
- jose_jwa_poly1305
- jose_jwa_sha3
- jose_jwa_unsupported
- jose_jwa_x25519
- jose_jwa_x448
- jose_jwa_xchacha20
- jose_jwa_xchacha20_poly1305
- jose_jwe
- jose_jwe_alg
- jose_jwe_alg_aes_kw
- jose_jwe_alg_c20p_kw
- jose_jwe_alg_dir
- jose_jwe_alg_ecdh_1pu
- jose_jwe_alg_ecdh_es
- jose_jwe_alg_ecdh_ss
- jose_jwe_alg_pbes2
- jose_jwe_alg_rsa
- jose_jwe_alg_xc20p_kw
- jose_jwe_enc
- jose_jwe_enc_aes
- jose_jwe_enc_c20p
- jose_jwe_enc_xc20p
- jose_jwe_zip
- jose_jwk
- jose_jwk_der
- jose_jwk_kty
- jose_jwk_kty_ec
- jose_jwk_kty_oct
- jose_jwk_kty_okp_ed25519
- jose_jwk_kty_okp_ed25519ph
- jose_jwk_kty_okp_ed448
- jose_jwk_kty_okp_ed448ph
- jose_jwk_kty_okp_x25519
- jose_jwk_kty_okp_x448
- jose_jwk_kty_rsa
- jose_jwk_oct
- jose_jwk_openssh_key
- jose_jwk_pem
- jose_jwk_set
- jose_jwk_use_enc
- jose_jwk_use_sig
- jose_jws
- jose_jws_alg
- jose_jws_alg_ecdsa
- jose_jws_alg_eddsa
- jose_jws_alg_hmac
- jose_jws_alg_none
- jose_jws_alg_poly1305
- jose_jws_alg_rsa_pkcs1_v1_5
- jose_jws_alg_rsa_pss
- jose_jwt
- jose_public_key
- jose_server
- jose_sha3
- jose_sha3_keccakf1600_driver
- jose_sha3_keccakf1600_nif
- jose_sha3_libdecaf
- jose_sha3_unsupported
- jose_sup
- jose_xchacha20_poly1305
- jose_xchacha20_poly1305_crypto
- jose_xchacha20_poly1305_libsodium
- jose_xchacha20_poly1305_unsupported
2024-05-23 05:42:29 +08:00
katana_code:
- ktn_code
- ktn_dodger
- ktn_io_string
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
khepri:
- khepri
- khepri_adv
- khepri_app
- khepri_cluster
- khepri_condition
- khepri_event_handler
- khepri_evf
- khepri_export_erlang
- khepri_import_export
- khepri_machine
- khepri_machine_v0
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
- khepri_path
- khepri_pattern_tree
- khepri_payload
- khepri_projection
- khepri_sproc
- khepri_sup
- khepri_tree
- khepri_tx
- khepri_tx_adv
- khepri_utils
khepri_mnesia_migration:
- khepri_mnesia_migration_app
- khepri_mnesia_migration_sup
- kmm_utils
- m2k_cluster_sync
- m2k_cluster_sync_sup
- m2k_export
- m2k_subscriber
- m2k_table_copy
- m2k_table_copy_sup
- m2k_table_copy_sup_sup
- mnesia_to_khepri
- mnesia_to_khepri_converter
- mnesia_to_khepri_example_converter
meck:
- meck
- meck_args_matcher
- meck_code
- meck_code_gen
- meck_cover
- meck_expect
- meck_history
- meck_matcher
- meck_proc
- meck_ret_spec
- meck_util
2024-02-21 12:02:24 +08:00
my_plugin:
- my_plugin
2024-01-03 16:28:36 +08:00
oauth2_client:
2024-05-17 17:24:33 +08:00
- jwt_helper
2024-01-03 16:28:36 +08:00
- oauth2_client
observer_cli:
- observer_cli
- observer_cli_application
- observer_cli_escriptize
- observer_cli_ets
- observer_cli_help
- observer_cli_inet
- observer_cli_lib
- observer_cli_mnesia
- observer_cli_plugin
- observer_cli_port
- observer_cli_process
- observer_cli_store
- observer_cli_system
osiris:
- osiris
- osiris_app
- osiris_bench
- osiris_bloom
- osiris_counters
- osiris_ets
- osiris_log
- osiris_log_shared
- osiris_replica
- osiris_replica_reader
- osiris_replica_reader_sup
- osiris_retention
- osiris_server_sup
- osiris_sup
- osiris_tracking
- osiris_util
- osiris_writer
prometheus:
- prometheus
- prometheus_boolean
- prometheus_buckets
- prometheus_collector
- prometheus_counter
- prometheus_format
- prometheus_gauge
- prometheus_histogram
- prometheus_http
- prometheus_instrumenter
- prometheus_metric
- prometheus_metric_spec
- prometheus_misc
- prometheus_mnesia
- prometheus_mnesia_collector
- prometheus_model
- prometheus_model_helpers
- prometheus_protobuf_format
- prometheus_quantile_summary
- prometheus_registry
- prometheus_summary
- prometheus_sup
- prometheus_test_instrumenter
- prometheus_text_format
- prometheus_time
- prometheus_vm_dist_collector
- prometheus_vm_memory_collector
- prometheus_vm_msacc_collector
- prometheus_vm_statistics_collector
- prometheus_vm_system_info_collector
proper:
- proper
- proper_arith
- proper_array
- proper_dict
- proper_erlang_abstract_code
- proper_fsm
- proper_gb_sets
- proper_gb_trees
- proper_gen
- proper_gen_next
- proper_orddict
- proper_ordsets
- proper_prop_remover
- proper_queue
- proper_sa
- proper_sets
- proper_shrink
- proper_statem
- proper_symb
- proper_target
- proper_transformer
- proper_types
- proper_typeserver
- proper_unicode
- proper_unused_imports_remover
- vararg
quantile_estimator:
- quantile
- quantile_estimator
ra:
- ra
- ra_app
- ra_aux
- ra_bench
- ra_counters
- ra_dbg
- ra_directory
- ra_env
2023-03-10 02:54:20 +08:00
- ra_ets_queue
- ra_file
- ra_file_handle
- ra_flru
- ra_leaderboard
- ra_lib
- ra_log
2023-03-10 02:54:20 +08:00
- ra_log_cache
- ra_log_ets
- ra_log_meta
- ra_log_pre_init
- ra_log_reader
- ra_log_segment
- ra_log_segment_writer
- ra_log_snapshot
- ra_log_sup
- ra_log_wal
- ra_log_wal_sup
- ra_machine
- ra_machine_ets
- ra_machine_simple
- ra_metrics_ets
- ra_monitors
- ra_server
- ra_server_proc
- ra_server_sup
- ra_server_sup_sup
- ra_snapshot
- ra_sup
- ra_system
- ra_system_recover
- ra_system_sup
- ra_systems_sup
rabbit:
- amqqueue
- background_gc
- code_server_cache
- gatherer
- gm
- internal_user
- lqueue
Message Containers (#5077) This PR implements an approach for a "protocol (data format) agnostic core" where the format of the message isn't converted at point of reception. Currently all non AMQP 0.9.1 originating messages are converted into a AMQP 0.9.1 flavoured basic_message record before sent to a queue. If the messages are then consumed by the originating protocol they are converted back from AMQP 0.9.1. For some protocols such as MQTT 3.1 this isn't too expensive as MQTT is mostly a fairly easily mapped subset of AMQP 0.9.1 but for others such as AMQP 1.0 the conversions are awkward and in some cases lossy even if consuming from the originating protocol. This PR instead wraps all incoming messages in their originating form into a generic, extensible message container type (mc). The container module exposes an API to get common message details such as size and various properties (ttl, priority etc) directly from the source data type. Each protocol needs to implement the mc behaviour such that when a message originating form one protocol is consumed by another protocol we convert it to the target protocol at that point. The message container also contains annotations, dead letter records and other meta data we need to record during the lifetime of a message. The original protocol message is never modified unless it is consumed. This includes conversion modules to and from amqp, amqpl (AMQP 0.9.1) and mqtt. COMMIT HISTORY: * Refactor away from using the delivery{} record In many places including exchange types. This should make it easier to move towards using a message container type instead of basic_message. Add mc module and move direct replies outside of exchange Lots of changes incl classic queues Implement stream support incl amqp conversions simplify mc state record move mc.erl mc dlx stuff recent history exchange Make tracking work But doesn't take a protocol agnostic approach as we just convert everything into AMQP legacy and back. Might be good enough for now. Tracing as a whole may want a bit of a re-vamp at some point. tidy make quorum queue peek work by legacy conversion dead lettering fixes dead lettering fixes CMQ fixes rabbit_trace type fixes fixes fix Fix classic queue props test assertion fix feature flag and backwards compat Enable message_container feature flag in some SUITEs Dialyzer fixes fixes fix test fixes Various Manually update a gazelle generated file until a gazelle enhancement can be made https://github.com/rabbitmq/rules_erlang/issues/185 Add message_containers_SUITE to bazel and regen bazel files with gazelle from rules_erlang@main Simplify essential proprty access Such as durable, ttl and priority by extracting them into annotations at message container init time. Move type to remove dependenc on amqp10 stuff in mc.erl mostly because I don't know how to make bazel do the right thing add more stuff Refine routing header stuff wip Cosmetics Do not use "maybe" as type name as "maybe" is a keyword since OTP 25 which makes Erlang LS complain. * Dedup death queue names * Fix function clause crashes Fix failing tests in the MQTT shared_SUITE: A classic queue message ID can be undefined as set in https://github.com/rabbitmq/rabbitmq-server/blob/fbe79ff47b4edbc0fd95457e623d6593161ad198/deps/rabbit/src/rabbit_classic_queue_index_v2.erl#L1048 Fix failing tests in the MQTT shared_SUITE-mixed: When feature flag message_containers is disabled, the message is not an #mc{} record, but a #basic_message{} record. * Fix is_utf8_no_null crash Prior to this commit, the function crashed if invalid UTF-8 was provided, e.g.: ``` 1> rabbit_misc:is_valid_shortstr(<<"😇"/utf16>>). ** exception error: no function clause matching rabbit_misc:is_utf8_no_null(<<216,61,222,7>>) (rabbit_misc.erl, line 1481) ``` * Implement mqtt mc behaviour For now via amqp translation. This is still work in progress, but the following SUITEs pass: ``` make -C deps/rabbitmq_mqtt ct-shared t=[mqtt,v5,cluster_size_1] FULL=1 make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_1] FULL=1 ``` * Shorten mc file names Module name length matters because for each persistent message the #mc{} record is persisted to disk. ``` 1> iolist_size(term_to_iovec({mc, rabbit_mc_amqp_legacy})). 30 2> iolist_size(term_to_iovec({mc, mc_amqpl})). 17 ``` This commit renames the mc modules: ``` ag -l rabbit_mc_amqp_legacy | xargs sed -i 's/rabbit_mc_amqp_legacy/mc_amqpl/g' ag -l rabbit_mc_amqp | xargs sed -i 's/rabbit_mc_amqp/mc_amqp/g' ag -l rabbit_mqtt_mc | xargs sed -i 's/rabbit_mqtt_mc/mc_mqtt/g' ``` * mc: make deaths an annotation + fixes * Fix mc_mqtt protocol_state callback * Fix test will_delay_node_restart ``` make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_3]:will_delay_node_restart FULL=1 ``` * Bazel run gazelle * mix format rabbitmqctl.ex * Ensure ttl annotation is refelected in amqp legacy protocol state * Fix id access in message store * Fix rabbit_message_interceptor_SUITE * dializer fixes * Fix rabbit:rabbit_message_interceptor_SUITE-mixed set_annotation/3 should not result in duplicate keys * Fix MQTT shared_SUITE-mixed Up to 3.12 non-MQTT publishes were always QoS 1 regardless of delivery_mode. https://github.com/rabbitmq/rabbitmq-server/blob/75a953ce286a10aca910c098805a4f545989af38/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl#L2075-L2076 From now on, non-MQTT publishes are QoS 1 if durable. This makes more sense. The MQTT plugin must send a #basic_message{} to an old node that does not understand message containers. * Field content of 'v1_0.data' can be binary Fix ``` bazel test //deps/rabbitmq_mqtt:shared_SUITE-mixed \ --test_env FOCUS="-group [mqtt,v4,cluster_size_1] -case trace" \ -t- --test_sharding_strategy=disabled ``` * Remove route/2 and implement route/3 for all exchange types. This removes the route/2 callback from rabbit_exchange_type and makes route/3 mandatory instead. This is a breaking change and will require all implementations of exchange types to update their code, however this is necessary anyway for them to correctly handle the mc type. stream filtering fixes * Translate directly from MQTT to AMQP 0.9.1 * handle undecoded properties in mc_compat amqpl: put clause in right order recover death deatails from amqp data * Replace callback init_amqp with convert_from * Fix return value of lists:keyfind/3 * Translate directly from AMQP 0.9.1 to MQTT * Fix MQTT payload size MQTT payload can be a list when converted from AMQP 0.9.1 for example First conversions tests Plus some other conversion related fixes. bazel bazel translate amqp 1.0 null to undefined mc: property/2 and correlation_id/message_id return type tagged values. To ensure we can support a variety of types better. The type type tags are AMQP 1.0 flavoured. fix death recovery mc_mqtt: impl new api Add callbacks to allow protocols to compact data before storage And make readable if needing to query things repeatedly. bazel fix * more decoding * tracking mixed versions compat * mc: flip default of `durable` annotation to save some data. Assuming most messages are durable and that in memory messages suffer less from persistence overhead it makes sense for a non existent `durable` annotation to mean durable=true. * mc conversion tests and tidy up * mc make x_header unstrict again * amqpl: death record fixes * bazel * amqp -> amqpl conversion test * Fix crash in mc_amqp:size/1 Body can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0/ ct-system t=java ``` on branch native-amqp. * Fix crash in lists:flatten/1 Data can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp. * Fix crash in rabbit_writer Running test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp resulted in the following crash: ``` crasher: initial call: rabbit_writer:enter_mainloop/2 pid: <0.711.0> registered_name: [] exception error: bad argument in function size/1 called as size([<<0>>,<<"Sw">>,[<<160,2>>,<<"hi">>]]) *** argument 1: not tuple or binary in call from rabbit_binary_generator:build_content_frames/7 (rabbit_binary_generator.erl, line 89) in call from rabbit_binary_generator:build_simple_content_frames/4 (rabbit_binary_generator.erl, line 61) in call from rabbit_writer:assemble_frames/5 (rabbit_writer.erl, line 334) in call from rabbit_writer:internal_send_command_async/3 (rabbit_writer.erl, line 365) in call from rabbit_writer:handle_message/2 (rabbit_writer.erl, line 265) in call from rabbit_writer:handle_message/3 (rabbit_writer.erl, line 232) in call from rabbit_writer:mainloop1/2 (rabbit_writer.erl, line 223) ``` because #content.payload_fragments_rev is currently supposed to be a flat list of binaries instead of being an iolist. This commit fixes this crash inefficiently by calling iolist_to_binary/1. A better solution would be to allow AMQP legacy's #content.payload_fragments_rev to be an iolist. * Add accidentally deleted line back * mc: optimise mc_amqp internal format By removint the outer records for message and delivery annotations as well as application properties and footers. * mc: optimis mc_amqp map_add by using upsert * mc: refactoring and bug fixes * mc_SUITE routingheader assertions * mc remove serialize/1 callback as only used by amqp * mc_amqp: avoid returning a nested list from protocol_state * test and bug fix * move infer_type to mc_util * mc fixes and additiona assertions * Support headers exchange routing for MQTT messages When a headers exchange is bound to the MQTT topic exchange, routing will be performend based on both MQTT topic (by the topic exchange) and MQTT User Property (by the headers exchange). This combines the best worlds of both MQTT 5.0 and AMQP 0.9.1 and enables powerful routing topologies. When the User Property contains the same name multiple times, only the last name (and value) will be considered by the headers exchange. * Fix crash when sending from stream to amqpl When publishing a message via the stream protocol and consuming it via AMQP 0.9.1, the following crash occurred prior to this commit: ``` crasher: initial call: rabbit_channel:init/1 pid: <0.818.0> registered_name: [] exception exit: {{badmatch,undefined}, [{rabbit_channel,handle_deliver0,4, [{file,"rabbit_channel.erl"}, {line,2728}]}, {lists,foldl,3,[{file,"lists.erl"},{line,1594}]}, {rabbit_channel,handle_cast,2, [{file,"rabbit_channel.erl"}, {line,728}]}, {gen_server2,handle_msg,2, [{file,"gen_server2.erl"},{line,1056}]}, {proc_lib,wake_up,3, [{file,"proc_lib.erl"},{line,251}]}]} ``` This commit first gives `mc:init/3` the chance to set exchange and routing_keys annotations. If not set, `rabbit_stream_queue` will set these annotations assuming the message was originally published via the stream protocol. * Support consistent hash exchange routing for MQTT 5.0 When a consistent hash exchange is bound to the MQTT topic exchange, MQTT 5.0 messages can be routed to queues consistently based on the Correlation-Data in the PUBLISH packet. * Convert MQTT 5.0 User Property * to AMQP 0.9.1 headers * from AMQP 0.9.1 headers * to AMQP 1.0 application properties and message annotations * from AMQP 1.0 application properties and message annotations * Make use of Annotations in mc_mqtt:protocol_state/2 mc_mqtt:protocol_state/2 includes Annotations as parameter. It's cleaner to make use of these Annotations when computing the protocol state instead of relying on the caller (rabbitmq_mqtt_processor) to compute the protocol state. * Enforce AMQP 0.9.1 field name length limit The AMQP 0.9.1 spec prohibits field names longer than 128 characters. Therefore, when converting AMQP 1.0 message annotations, application properties or MQTT 5.0 User Property to AMQP 0.9.1 headers, drop any names longer than 128 characters. * Fix type specs Apply feedback from Michael Davis Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * Add mc_mqtt unit test suite Implement mc_mqtt:x_header/2 * Translate indicator that payload is UTF-8 encoded when converting between MQTT 5.0 and AMQP 1.0 * Translate single amqp-value section from AMQP 1.0 to MQTT Convert to a text representation, if possible, and indicate to MQTT client that the payload is UTF-8 encoded. This way, the MQTT client will be able to parse the payload. If conversion to text representation is not possible, encode the payload using the AMQP 1.0 type system and indiate the encoding via Content-Type message/vnd.rabbitmq.amqp. This Content-Type is not registered. Type "message" makes sense since it's a message. Vendor tree "vnd.rabbitmq.amqp" makes sense since merely subtype "amqp" is not registered. * Fix payload conversion * Translate Response Topic between MQTT and AMQP Translate MQTT 5.0 Response Topic to AMQP 1.0 reply-to address and vice versa. The Response Topic must be a UTF-8 encoded string. This commit re-uses the already defined RabbitMQ target addresses: ``` "/topic/" RK Publish to amq.topic with routing key RK "/exchange/" X "/" RK Publish to exchange X with routing key RK ``` By default, the MQTT topic exchange is configure dto be amq.topic using the 1st target address. When an operator modifies the mqtt.exchange, the 2nd target address is used. * Apply PR feedback and fix formatting Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * tidy up * Add MQTT message_containers test * consistent hash exchange: avoid amqp legacy conversion When hashing on a header value. * Avoid converting to amqp legacy when using exchange federation * Fix test flake * test and dialyzer fixes * dialyzer fix * Add MQTT protocol interoperability tests Test receiving from and sending to MQTT 5.0 and * AMQP 0.9.1 * AMQP 1.0 * STOMP * Streams * Regenerate portions of deps/rabbit/app.bzl with gazelle I'm not exactly sure how this happened, but gazell seems to have been run with an older version of the rules_erlang gazelle extension at some point. This caused generation of a structure that is no longer used. This commit updates the structure to the current pattern. * mc: refactoring * mc_amqpl: handle delivery annotations Just in case they are included. Also use iolist_to_iovec to create flat list of binaries when converting from amqp with amqp encoded payload. --------- Co-authored-by: David Ansari <david.ansari@gmx.de> Co-authored-by: Michael Davis <mcarsondavis@gmail.com> Co-authored-by: Rin Kuryloski <kuryloskip@vmware.com>
2023-08-31 18:27:13 +08:00
- mc
- mc_amqp
- mc_amqpl
- mc_compat
- mc_util
- mirrored_supervisor
- mirrored_supervisor_sups
- pg_local
- pid_recomposition
- rabbit
- rabbit_access_control
- rabbit_alarm
Support AMQP 1.0 natively ## What Similar to Native MQTT in #5895, this commits implements Native AMQP 1.0. By "native", we mean do not proxy via AMQP 0.9.1 anymore. ## Why Native AMQP 1.0 comes with the following major benefits: 1. Similar to Native MQTT, this commit provides better throughput, latency, scalability, and resource usage for AMQP 1.0. See https://blog.rabbitmq.com/posts/2023/03/native-mqtt for native MQTT improvements. See further below for some benchmarks. 2. Since AMQP 1.0 is not limited anymore by the AMQP 0.9.1 protocol, this commit allows implementing more AMQP 1.0 features in the future. Some features are already implemented in this commit (see next section). 3. Simpler, better understandable, and more maintainable code. Native AMQP 1.0 as implemented in this commit has the following major benefits compared to AMQP 0.9.1: 4. Memory and disk alarms will only stop accepting incoming TRANSFER frames. New connections can still be created to consume from RabbitMQ to empty queues. 5. Due to 4. no need anymore for separate connections for publishers and consumers as we currently recommended for AMQP 0.9.1. which potentially halves the number of physical TCP connections. 6. When a single connection sends to multiple target queues, a single slow target queue won't block the entire connection. Publisher can still send data quickly to all other target queues. 7. A publisher can request whether it wants publisher confirmation on a per-message basis. In AMQP 0.9.1 publisher confirms are configured per channel only. 8. Consumers can change their "prefetch count" dynamically which isn't possible in our AMQP 0.9.1 implementation. See #10174 9. AMQP 1.0 is an extensible protocol This commit also fixes dozens of bugs present in the AMQP 1.0 plugin in RabbitMQ 3.x - most of which cannot be backported due to the complexity and limitations of the old 3.x implementation. This commit contains breaking changes and is therefore targeted for RabbitMQ 4.0. ## Implementation details 1. Breaking change: With Native AMQP, the behaviour of ``` Convert AMQP 0.9.1 message headers to application properties for an AMQP 1.0 consumer amqp1_0.convert_amqp091_headers_to_app_props = false | true (default false) Convert AMQP 1.0 Application Properties to AMQP 0.9.1 headers amqp1_0.convert_app_props_to_amqp091_headers = false | true (default false) ``` will break because we always convert according to the message container conversions. For example, AMQP 0.9.1 x-headers will go into message-annotations instead of application properties. Also, `false` won’t be respected since we always convert the headers with message containers. 2. Remove rabbit_queue_collector rabbit_queue_collector is responsible for synchronously deleting exclusive queues. Since the AMQP 1.0 plugin never creates exclusive queues, rabbit_queue_collector doesn't need to be started in the first place. This will save 1 Erlang process per AMQP 1.0 connection. 3. 7 processes per connection + 1 process per session in this commit instead of 7 processes per connection + 15 processes per session in 3.x Supervision hierarchy got re-designed. 4. Use 1 writer process per AMQP 1.0 connection AMQP 0.9.1 uses a separate rabbit_writer Erlang process per AMQP 0.9.1 channel. Prior to this commit, AMQP 1.0 used a separate rabbit_amqp1_0_writer process per AMQP 1.0 session. Advantage of single writer proc per session (prior to this commit): * High parallelism for serialising packets if multiple sessions within a connection write heavily at the same time. This commit uses a single writer process per AMQP 1.0 connection that is shared across all AMQP 1.0 sessions. Advantages of single writer proc per connection (this commit): * Lower memory usage with hundreds of thousands of AMQP 1.0 sessions * Less TCP and IP header overhead given that the single writer process can accumulate across all sessions bytes before flushing the socket. In other words, this commit decides that a reader / writer process pair per AMQP 1.0 connection is good enough for bi-directional TRANSFER flows. Having a writer per session is too heavy. We still ensure high throughput by having separate reader, writer, and session processes. 5. Transform rabbit_amqp1_0_writer into gen_server Why: Prior to this commit, when clicking on the AMQP 1.0 writer process in observer, the process crashed. Instead of handling all these debug messages of the sys module, it's better to implement a gen_server. There is no advantage of using a special OTP process over gen_server for the AMQP 1.0 writer. gen_server also provides cleaner format status output. How: Message callbacks return a timeout of 0. After all messages in the inbox are processed, the timeout message is handled by flushing any pending bytes. 6. Remove stats timer from writer AMQP 1.0 connections haven't emitted any stats previously. 7. When there are contiguous queue confirmations in the session process mailbox, batch them. When the confirmations are sent to the publisher, a single DISPOSITION frame is sent for contiguously confirmed delivery IDs. This approach should be good enough. However it's sub optimal in scenarios where contiguous delivery IDs that need confirmations are rare, for example: * There are multiple links in the session with different sender settlement modes and sender publishes across these links interleaved. * sender settlement mode is mixed and sender publishes interleaved settled and unsettled TRANSFERs. 8. Introduce credit API v2 Why: The AMQP 0.9.1 credit extension which is to be removed in 4.0 was poorly designed since basic.credit is a synchronous call into the queue process blocking the entire AMQP 1.0 session process. How: Change the interactions between queue clients and queue server implementations: * Clients only request a credit reply if the FLOW's `echo` field is set * Include all link flow control state held by the queue process into a new credit_reply queue event: * `available` after the queue sends any deliveries * `link-credit` after the queue sends any deliveries * `drain` which allows us to combine the old queue events send_credit_reply and send_drained into a single new queue event credit_reply. * Include the consumer tag into the credit_reply queue event such that the AMQP 1.0 session process can process any credit replies asynchronously. Link flow control state `delivery-count` also moves to the queue processes. The new interactions are hidden behind feature flag credit_api_v2 to allow for rolling upgrades from 3.13 to 4.0. 9. Use serial number arithmetic in quorum queues and session process. 10. Completely bypass the rabbit_limiter module for AMQP 1.0 flow control. The goal is to eventually remove the rabbit_limiter module in 4.0 since AMQP 0.9.1 global QoS will be unsupported in 4.0. This commit lifts the AMQP 1.0 link flow control logic out of rabbit_limiter into rabbit_queue_consumers. 11. Fix credit bug for streams: AMQP 1.0 settlements shouldn't top up link credit, only FLOW frames should top up link credit. 12. Allow sender settle mode unsettled for streams since AMQP 1.0 acknowledgements to streams are no-ops (currently). 13. Fix AMQP 1.0 client bugs Auto renewing credits should not be related to settling TRANSFERs. Remove field link_credit_unsettled as it was wrong and confusing. Prior to this commit auto renewal did not work when the sender uses sender settlement mode settled. 14. Fix AMQP 1.0 client bugs The wrong outdated Link was passed to function auto_flow/2 15. Use osiris chunk iterator Only hold messages of uncompressed sub batches in memory if consumer doesn't have sufficient credits. Compressed sub batches are skipped for non Stream protocol consumers. 16. Fix incoming link flow control Always use confirms between AMQP 1.0 queue clients and queue servers. As already done internally by rabbit_fifo_client and rabbit_stream_queue, use confirms for classic queues as well. 17. Include link handle into correlation when publishing messages to target queues such that session process can correlate confirms from target queues to incoming links. 18. Only grant more credits to publishers if publisher hasn't sufficient credits anymore and there are not too many unconfirmed messages on the link. 19. Completely ignore `block` and `unblock` queue actions and RabbitMQ credit flow between classic queue process and session process. 20. Link flow control is independent between links. A client can refer to a queue or to an exchange with multiple dynamically added target queues. Multiple incoming links can also fan in to the same queue. However the link topology looks like, this commit ensures that each link is only granted more credits if that link isn't overloaded. 21. A connection or a session can send to many different queues. In AMQP 0.9.1, a single slow queue will lead to the entire channel, and then entire connection being blocked. This commit makes sure that a single slow queue from one link won't slow down sending on other links. For example, having link A sending to a local classic queue and link B sending to 5 replica quorum queue, link B will naturally grant credits slower than link A. So, despite the quorum queue being slower in confirming messages, the same AMQP 1.0 connection and session can still pump data very fast into the classic queue. 22. If cluster wide memory or disk alarm occurs. Each session sends a FLOW with incoming-window to 0 to sending client. If sending clients don’t obey, force disconnect the client. If cluster wide memory alarm clears: Each session resumes with a FLOW defaulting to initial incoming-window. 23. All operations apart of publishing TRANSFERS to RabbitMQ can continue during cluster wide alarms, specifically, attaching consumers and consuming, i.e. emptying queues. There is no need for separate AMQP 1.0 connections for publishers and consumers as recommended in our AMQP 0.9.1 implementation. 24. Flow control summary: * If queue becomes bottleneck, that’s solved by slowing down individual sending links (AMQP 1.0 link flow control). * If session becomes bottleneck (more unlikely), that’s solved by AMQP 1.0 session flow control. * If connection becomes bottleneck, it naturally won’t read fast enough from the socket causing TCP backpressure being applied. Nowhere will RabbitMQ internal credit based flow control (i.e. module credit_flow) be used on the incoming AMQP 1.0 message path. 25. Register AMQP sessions Prefer local-only pg over our custom pg_local implementation as pg is a better process group implementation than pg_local. pg_local was identified as bottleneck in tests where many MQTT clients were disconnected at once. 26. Start a local-only pg when Rabbit boots: > A scope can be kept local-only by using a scope name that is unique cluster-wide, e.g. the node name: > pg:start_link(node()). Register AMQP 1.0 connections and sessions with pg. In future we should remove pg_local and instead use the new local-only pg for all registered processes such as AMQP 0.9.1 connections and channels. 27. Requeue messages if link detached Although the spec allows to settle delivery IDs on detached links, RabbitMQ does not respect the 'closed' field of the DETACH frame and therefore handles every DETACH frame as closed. Since the link is closed, we expect every outstanding delivery to be requeued. In addition to consumer cancellation, detaching a link therefore causes in flight deliveries to be requeued. Note that this behaviour is different from merely consumer cancellation in AMQP 0.9.1: "After a consumer is cancelled there will be no future deliveries dispatched to it. Note that there can still be "in flight" deliveries dispatched previously. Cancelling a consumer will neither discard nor requeue them." [https://www.rabbitmq.com/consumers.html#unsubscribing] An AMQP receiver can first drain, and then detach to prevent "in flight" deliveries 28. Init AMQP session with BEGIN frame Similar to how there can't be an MQTT processor without a CONNECT frame, there can't be an AMQP session without a BEGIN frame. This allows having strict dialyzer types for session flow control fields (i.e. not allowing 'undefined'). 29. Move serial_number to AMQP 1.0 common lib such that it can be used by both AMQP 1.0 server and client 30. Fix AMQP client to do serial number arithmetic. 31. AMQP client: Differentiate between delivery-id and transfer-id for better understandability. 32. Fix link flow control in classic queues This commit fixes ``` java -jar target/perf-test.jar -ad false -f persistent -u cq -c 3000 -C 1000000 -y 0 ``` followed by ``` ./omq -x 0 amqp -T /queue/cq -D 1000000 --amqp-consumer-credits 2 ``` Prior to this commit, (and on RabbitMQ 3.x) the consuming would halt after around 8 - 10,000 messages. The bug was that in flight messages from classic queue process to session process were not taken into account when topping up credit to the classic queue process. Fixes #2597 The solution to this bug (and a much cleaner design anyway independent of this bug) is that queues should hold all link flow control state including the delivery-count. Hence, when credit API v2 is used the delivery-count will be held by the classic queue process, quorum queue process, and stream queue client instead of managing the delivery-count in the session. 33. The double level crediting between (a) session process and rabbit_fifo_client, and (b) rabbit_fifo_client and rabbit_fifo was removed. Therefore, instead of managing 3 separate delivery-counts (i. session, ii. rabbit_fifo_client, iii. rabbit_fifo), only 1 delivery-count is used in rabbit_fifo. This is a big simplification. 34. This commit fixes quorum queues without bumping the machine version nor introducing new rabbit_fifo commands. Whether credit API v2 is used is solely determined at link attachment time depending on whether feature flag credit_api_v2 is enabled. Even when that feature flag will be enabled later on, this link will keep using credit API v1 until detached (or the node is shut down). Eventually, after feature flag credit_api_v2 has been enabled and a subsequent rolling upgrade, all links will use credit API v2. This approach is safe and simple. The 2 alternatives to move delivery-count from the session process to the queue processes would have been: i. Explicit feature flag credit_api_v2 migration function * Can use a gen_server:call and only finish migration once all delivery-counts were migrated. Cons: * Extra new message format just for migration is required. * Risky as migration will fail if a target queue doesn’t reply. ii. Session always includes DeliveryCountSnd when crediting to the queue: Cons: * 2 delivery counts will be hold simultaneously in session proc and queue proc; could be solved by deleting the session proc’s delivery-count for credit-reply * What happens if the receiver doesn’t provide credit for a very long time? Is that a problem? 35. Support stream filtering in AMQP 1.0 (by @acogoluegnes) Use the x-stream-filter-value message annotation to carry the filter value in a published message. Use the rabbitmq:stream-filter and rabbitmq:stream-match-unfiltered filters when creating a receiver that wants to filter out messages from a stream. 36. Remove credit extension from AMQP 0.9.1 client 37. Support maintenance mode closing AMQP 1.0 connections. 38. Remove AMQP 0.9.1 client dependency from AMQP 1.0 implementation. 39. Move AMQP 1.0 plugin to the core. AMQP 1.0 is enabled by default. The old rabbitmq_amqp1_0 plugin will be kept as a no-op plugin to prevent deployment tools from failing that execute: ``` rabbitmq-plugins enable rabbitmq_amqp1_0 rabbitmq-plugins disable rabbitmq_amqp1_0 ``` 40. Breaking change: Remove CLI command `rabbitmqctl list_amqp10_connections`. Instead, list both AMQP 0.9.1 and AMQP 1.0 connections in `list_connections`: ``` rabbitmqctl list_connections protocol Listing connections ... protocol {1, 0} {0,9,1} ``` ## Benchmarks ### Throughput & Latency Setup: * Single node Ubuntu 22.04 * Erlang 26.1.1 Start RabbitMQ: ``` make run-broker PLUGINS="rabbitmq_management rabbitmq_amqp1_0" FULL=1 RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+S 3" ``` Predeclare durable classic queue cq1, durable quorum queue qq1, durable stream queue sq1. Start client: https://github.com/ssorj/quiver https://hub.docker.com/r/ssorj/quiver/tags (digest 453a2aceda64) ``` docker run -it --rm --add-host host.docker.internal:host-gateway ssorj/quiver:latest bash-5.1# quiver --version quiver 0.4.0-SNAPSHOT ``` 1. Classic queue ``` quiver //host.docker.internal//amq/queue/cq1 --durable --count 1m --duration 10m --body-size 12 --credit 1000 ``` This commit: ``` Count ............................................. 1,000,000 messages Duration ............................................... 73.8 seconds Sender rate .......................................... 13,548 messages/s Receiver rate ........................................ 13,547 messages/s End-to-end rate ...................................... 13,547 messages/s Latencies by percentile: 0% ........ 0 ms 90.00% ........ 9 ms 25% ........ 2 ms 99.00% ....... 14 ms 50% ........ 4 ms 99.90% ....... 17 ms 100% ....... 26 ms 99.99% ....... 24 ms ``` RabbitMQ 3.x (main branch as of 30 January 2024): ``` ---------------------- Sender ----------------------- --------------------- Receiver ---------------------- -------- Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Lat [ms] ----------------------------------------------------- ----------------------------------------------------- -------- 2.1 130,814 65,342 6 73.6 2.1 3,217 1,607 0 8.0 511 4.1 163,580 16,367 2 74.1 4.1 3,217 0 0 8.0 0 6.1 229,114 32,767 3 74.1 6.1 3,217 0 0 8.0 0 8.1 261,880 16,367 2 74.1 8.1 67,874 32,296 8 8.2 7,662 10.1 294,646 16,367 2 74.1 10.1 67,874 0 0 8.2 0 12.1 360,180 32,734 3 74.1 12.1 67,874 0 0 8.2 0 14.1 392,946 16,367 3 74.1 14.1 68,604 365 0 8.2 12,147 16.1 458,480 32,734 3 74.1 16.1 68,604 0 0 8.2 0 18.1 491,246 16,367 2 74.1 18.1 68,604 0 0 8.2 0 20.1 556,780 32,767 4 74.1 20.1 68,604 0 0 8.2 0 22.1 589,546 16,375 2 74.1 22.1 68,604 0 0 8.2 0 receiver timed out 24.1 622,312 16,367 2 74.1 24.1 68,604 0 0 8.2 0 quiver: error: PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/cq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-otujr23y' returned non-zero exit status 1. Traceback (most recent call last): File "/usr/local/lib/quiver/python/quiver/pair.py", line 144, in run _plano.wait(receiver, check=True) File "/usr/local/lib/quiver/python/plano/main.py", line 1243, in wait raise PlanoProcessError(proc) plano.main.PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/cq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-otujr23y' returned non-zero exit status 1. ``` 2. Quorum queue: ``` quiver //host.docker.internal//amq/queue/qq1 --durable --count 1m --duration 10m --body-size 12 --credit 1000 ``` This commit: ``` Count ............................................. 1,000,000 messages Duration .............................................. 101.4 seconds Sender rate ........................................... 9,867 messages/s Receiver rate ......................................... 9,868 messages/s End-to-end rate ....................................... 9,865 messages/s Latencies by percentile: 0% ....... 11 ms 90.00% ....... 23 ms 25% ....... 15 ms 99.00% ....... 28 ms 50% ....... 18 ms 99.90% ....... 33 ms 100% ....... 49 ms 99.99% ....... 47 ms ``` RabbitMQ 3.x: ``` ---------------------- Sender ----------------------- --------------------- Receiver ---------------------- -------- Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Lat [ms] ----------------------------------------------------- ----------------------------------------------------- -------- 2.1 130,814 65,342 9 69.9 2.1 18,430 9,206 5 7.6 1,221 4.1 163,580 16,375 5 70.2 4.1 18,867 218 0 7.6 2,168 6.1 229,114 32,767 6 70.2 6.1 18,867 0 0 7.6 0 8.1 294,648 32,734 7 70.2 8.1 18,867 0 0 7.6 0 10.1 360,182 32,734 6 70.2 10.1 18,867 0 0 7.6 0 12.1 425,716 32,767 6 70.2 12.1 18,867 0 0 7.6 0 receiver timed out 14.1 458,482 16,367 5 70.2 14.1 18,867 0 0 7.6 0 quiver: error: PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/qq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-b1gcup43' returned non-zero exit status 1. Traceback (most recent call last): File "/usr/local/lib/quiver/python/quiver/pair.py", line 144, in run _plano.wait(receiver, check=True) File "/usr/local/lib/quiver/python/plano/main.py", line 1243, in wait raise PlanoProcessError(proc) plano.main.PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/qq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-b1gcup43' returned non-zero exit status 1. ``` 3. Stream: ``` quiver-arrow send //host.docker.internal//amq/queue/sq1 --durable --count 1m -d 10m --summary --verbose ``` This commit: ``` Count ............................................. 1,000,000 messages Duration ................................................ 8.7 seconds Message rate ........................................ 115,154 messages/s ``` RabbitMQ 3.x: ``` Count ............................................. 1,000,000 messages Duration ............................................... 21.2 seconds Message rate ......................................... 47,232 messages/s ``` ### Memory usage Start RabbitMQ: ``` ERL_MAX_PORTS=3000000 RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+P 3000000 +S 6" make run-broker PLUGINS="rabbitmq_amqp1_0" FULL=1 RABBITMQ_CONFIG_FILE="rabbitmq.conf" ``` ``` /bin/cat rabbitmq.conf tcp_listen_options.sndbuf = 2048 tcp_listen_options.recbuf = 2048 vm_memory_high_watermark.relative = 0.95 vm_memory_high_watermark_paging_ratio = 0.95 loopback_users = none ``` Create 50k connections with 2 sessions per connection, i.e. 100k session in total: ```go package main import ( "context" "log" "time" "github.com/Azure/go-amqp" ) func main() { for i := 0; i < 50000; i++ { conn, err := amqp.Dial(context.TODO(), "amqp://nuc", &amqp.ConnOptions{SASLType: amqp.SASLTypeAnonymous()}) if err != nil { log.Fatal("dialing AMQP server:", err) } _, err = conn.NewSession(context.TODO(), nil) if err != nil { log.Fatal("creating AMQP session:", err) } _, err = conn.NewSession(context.TODO(), nil) if err != nil { log.Fatal("creating AMQP session:", err) } } log.Println("opened all connections") time.Sleep(5 * time.Hour) } ``` This commit: ``` erlang:memory(). [{total,4586376480}, {processes,4025898504}, {processes_used,4025871040}, {system,560477976}, {atom,1048841}, {atom_used,1042841}, {binary,233228608}, {code,21449982}, {ets,108560464}] erlang:system_info(process_count). 450289 ``` 7 procs per connection + 1 proc per session. (7 + 2*1) * 50,000 = 450,000 procs RabbitMQ 3.x: ``` erlang:memory(). [{total,15168232704}, {processes,14044779256}, {processes_used,14044755120}, {system,1123453448}, {atom,1057033}, {atom_used,1052587}, {binary,236381264}, {code,21790238}, {ets,391423744}] erlang:system_info(process_count). 1850309 ``` 7 procs per connection + 15 per session (7 + 2*15) * 50,000 = 1,850,000 procs 50k connections + 100k session require with this commit: 4.5 GB in RabbitMQ 3.x: 15 GB ## Future work 1. More efficient parser and serializer 2. TODO in mc_amqp: Do not store the parsed message on disk. 3. Implement both AMQP HTTP extension and AMQP management extension to allow AMQP clients to create RabbitMQ objects (queues, exchanges, ...).
2023-07-21 18:29:07 +08:00
- rabbit_amqp1_0
Enable AMQP 1.0 clients to manage topologies ## What? * Allow AMQP 1.0 clients to dynamically create and delete RabbitMQ topologies (exchanges, queues, bindings). * Provide an Erlang AMQP 1.0 client that manages topologies. ## Why? Today, RabbitMQ topologies can be created via: * [Management HTTP API](https://www.rabbitmq.com/docs/management#http-api) (including Management UI and [messaging-topology-operator](https://github.com/rabbitmq/messaging-topology-operator)) * [Definition Import](https://www.rabbitmq.com/docs/definitions#import) * AMQP 0.9.1 clients Up to RabbitMQ 3.13 the RabbitMQ AMQP 1.0 plugin auto creates queues and bindings depending on the terminus [address format](https://github.com/rabbitmq/rabbitmq-server/tree/v3.13.x/deps/rabbitmq_amqp1_0#routing-and-addressing). Such implicit creation of topologies is limiting and obscure. For some address formats, queues will be created, but not deleted. Some of RabbitMQ's success is due to its flexible routing topologies that AMQP 0.9.1 clients can create and delete dynamically. This commit allows dynamic management of topologies for AMQP 1.0 clients. This commit builds on top of Native AMQP 1.0 (PR #9022) and will be available in RabbitMQ 4.0. ## How? This commits adds the following management operations for AMQP 1.0 clients: * declare queue * delete queue * purge queue * bind queue to exchange * unbind queue from exchange * declare exchange * delete exchange * bind exchange to exchange * unbind exchange from exchange Hence, at least the AMQP 0.9.1 management operations are supported for AMQP 1.0 clients. In addition the operation * get queue is provided which - similar to `declare queue` - returns queue information including the current leader and replicas. This allows clients to publish or consume locally on the node that hosts the queue. Compared to AMQP 0.9.1 whose commands and command fields are fixed, the new AMQP Management API is extensible: New operations and new fields can easily be added in the future. There are different design options how management operations could be supported for AMQP 1.0 clients: 1. Use a special exchange type as done in https://github.com/rabbitmq/rabbitmq-management-exchange This has the advantage that any protocol client (e.g. also STOMP clients) could dynamically manage topologies. However, a special exchange type is the wrong abstraction. 2. Clients could send "special" messages with special headers that the broker interprets. This commit decided for a variation of the 2nd option using a more standardized way by re-using a subest of the following latest AMQP 1.0 extension specifications: * [AMQP Request-Response Messaging with Link Pairing Version 1.0 - Committee Specification 01](https://docs.oasis-open.org/amqp/linkpair/v1.0/cs01/linkpair-v1.0-cs01.html) (February 2021) * [HTTP Semantics and Content over AMQP Version 1.0 - Working Draft 06](https://groups.oasis-open.org/higherlogic/ws/public/document?document_id=65571) (July 2019) * [AMQP Management Version 1.0 - Working Draft 16](https://groups.oasis-open.org/higherlogic/ws/public/document?document_id=65575) (July 2019) An important goal is to keep the interaction between AMQP 1.0 client and RabbitMQ simple to increase usage, development and adoptability of future RabbitMQ AMQP 1.0 client library wrappers. The AMQP 1.0 client has to create a link pair to the special `/management` node. This allows the client to send and receive from the management node. Similar to AMQP 0.9.1, there is no need for a reply queue since the reply will be sent directly to the client. Requests and responses are modelled via HTTP, but sent via AMQP using the `HTTP Semantics and Content over AMQP` extension (henceforth `HTTP over AMQP` extension). This commit tries to follow the `HTTP over AMQP` extension as much as possible but deviates where this draft spec doesn't make sense. The projected mode §4.1 is used as opposed to tunneled mode §4.2. A named relay `/management` is used (§6.3) where the message field `to` is the URL. Deviations are * §3.1 mandates that URIs are not encoded in an AMQP message. However, we percent encode URIs in the AMQP message. Otherwise there is for example no way to distinguish a `/` in a queue name from the URI path separator `/`. * §4.1.4 mandates a data section. This commit uses an amqp-value section as it's a better fit given that the content is AMQP encoded data. Using an HTTP API allows for a common well understood interface and future extensibility. Instead of re-using the current RabbitMQ HTTP API, this commit uses a new HTTP API (let's call it v2) which could be used as a future API for plain HTTP clients. ### HTTP API v1 The current HTTP API (let's call it v1) is **not** used since v1 comes with a couple of weaknesses: 1. Deep level of nesting becomes confusing and difficult to manage. Examples of deep nesting in v1: ``` /api/bindings/vhost/e/source/e/destination/props /api/bindings/vhost/e/exchange/q/queue/props ``` 2. Redundant endpoints returning the same resources v1 has 9 endpoints to list binding(s): ``` /api/exchanges/vhost/name/bindings/source /api/exchanges/vhost/name/bindings/destination /api/queues/vhost/name/bindings /api/bindings /api/bindings/vhost /api/bindings/vhost/e/exchange/q/queue /api/bindings/vhost/e/exchange/q/queue/props /api/bindings/vhost/e/source/e/destination /api/bindings/vhost/e/source/e/destination/props ``` 3. Verbs in path names Path names should be nouns instead. v1 contains verbs: ``` /api/queues/vhost/name/get /api/exchanges/vhost/name/publish ``` ### AMQP Management extension Only few aspects of the AMQP Management extension are used. The central idea of the AMQP management spec is **dynamic discovery** such that broker independent AMQP 1.0 clients can discover objects, types, operations, and HTTP endpoints of specific brokers. In fact, clients are only conformant if: > All request addresses are dynamically discovered starting from the discovery document. > A requesting container MUST NOT use fixed assumptions about the addressing structure of the management API. While this is a nice and powerful idea, no AMQP 1.0 client and no AMQP 1.0 server implement the latest AMQP 1.0 management spec from 2019, partly presumably due to its complexity. Therefore, the idea of such dynamic discovery has failed to be implemented in practice. The AMQP management spec mandates that the management endpoint returns a discovery document containing broker specific collections, types, configuration, and operations including their endpoints. The API endpoints of the AMQP management spec are therefore all designed around dynamic discovery. For example, to create either a queue or an exchange, the client has to ``` POST /$management/entities ``` which shows that the entities collection acts as a generic factory, see section 2.2. The server will then create the resource and reply with a location header containing a URI pointing to the resource. For RabbitMQ, we don’t need such a generic factory to create queues or exchanges. To list bindings for a queue Q1, the spec suggests ``` GET /$management/Queues/Q1/$management/entities ``` which again shows the generic entities endpoint as well as a `$management` endpoint under Q1 to allow a queue to return a discovery document. For RabbitMQ, we don’t need such generic endpoints and discovery documents. Given we aim for our own thin RabbitMQ AMQP 1.0 client wrapper libraries which expose the RabbitMQ model to the developer, we can directly use fixed HTTP endpoint assumptions in our RabbitMQ specific libraries. This is by far simpler than using the dynamic endpoints of the management spec. Simplicity leads to higher adoption and enables more developers to write RabbitMQ AMQP 1.0 client library wrappers. The AMQP Management extension also suffers from deep level of nesting in paths Examples: ``` /$management/Queues/Q1/$management/entities /$management/Queues/Q1/Bindings/Binding1 ``` as well as verbs in path names: Section 7.1.4 suggests using verbs in path names, for example “purge”, due to the dynamic operations discovery document. ### HTTP API v2 This commit introduces a new HTTP API v2 following best practices. It could serve as a future API for plain HTTP clients. This commit and RabbitMQ 4.0 will only implement a minimal set of HTTP API v2 endpoints and only for HTTP over AMQP. In other words, the existing HTTP API v1 Cowboy handlers will continue to be used for all plain HTTP requests in RabbitMQ 4.0 and will remain untouched for RabbitMQ 4.0. Over time, after 4.0 shipped, we could ship a pure HTTP API implementation for HTTP API v2. Hence, the new HTTP API v2 endpoints for HTTP over AMQP should be designed such that they can be re-used in the future for a pure HTTP implementation. The minimal set of endpoints for RabbitMQ 4.0 are: `` GET / PUT / DELETE /vhosts/:vhost/queues/:queue ``` read, create, delete a queue ``` DELETE /vhosts/:vhost/queues/:queue/messages ``` purges a queue ``` GET / DELETE /vhosts/:vhost/bindings/:binding ``` read, delete bindings where `:binding` is a binding ID of the following path segment: ``` src=e1;dstq=q2;key=my-key;args= ``` Binding arguments `args` has an empty value by default, i.e. there are no binding arguments. If the binding includes binding arguments, `args` will be an Erlang portable term hash provided by the server similar to what’s provided in HTTP API v1 today. Alternatively, we could use an arguments scheme of: ``` args=k1,utf8,v1&k2,uint,3 ``` However, such a scheme leads to long URIs when there are many binding arguments. Note that it’s perfectly fine for URI producing applications to include URI reserved characters `=` / `;` / `,` / `$` in a path segment. To create a binding, the client therefore needs to POST to a bindings factory URI: ``` POST /vhosts/:vhost/bindings ``` To list all bindings between a source exchange e1 and destination exchange e2 with binding key k1: ``` GET /vhosts/:vhost/bindings?src=e1&dste=e2&key=k1 ``` This endpoint will be called by the RabbitMQ AMQP 1.0 client library to unbind a binding with non-empty binding arguments to get the binding ID before invoking a ``` DELETE /vhosts/:vhost/bindings/:binding ``` In future, after RabbitMQ 4.0 shipped, new API endpoints could be added. The following is up for discussion and is only meant to show the clean and simple design of HTTP API v2. Bindings endpoint can be queried as follows: to list all bindings for a given source exchange e1: ``` GET /vhosts/:vhost/bindings?src=e1 ``` to list all bindings for a given destination queue q1: ``` GET /vhosts/:vhost/bindings?dstq=q1 ``` to list all bindings between a source exchange e1 and destination queue q1: ``` GET /vhosts/:vhost/bindings?src=e1&dstq=q1 ``` multiple bindings between source exchange e1 and destination queue q1 could be deleted at once as follows: ``` DELETE /vhosts/:vhost/bindings?src=e1&dstq=q1 ``` GET could be supported globally across all vhosts: ``` /exchanges /queues /bindings ``` Publish a message: ``` POST /vhosts/:vhost/queues/:queue/messages ``` Consume or peek a message (depending on query parameters): ``` GET /vhosts/:vhost/queues/:queue/messages ``` Note that the AMQP 1.0 client omits the `/vhost/:vhost` path prefix. Since an AMQP connection belongs to a single vhost, there is no need to additionally include the vhost in every HTTP request. Pros of HTTP API v2: 1. Low level of nesting Queues, exchanges, bindings are top level entities directly under vhosts. Although the HTTP API doesn’t have to reflect how resources are stored in the database, v2 does nicely reflect the Khepri tree structure. 2. Nouns instead of verbs HTTP API v2 is very simple to read and understand as shown by ``` POST /vhosts/:vhost/queues/:queue/messages to post messages, i.e. publish to a queue. GET /vhosts/:vhost/queues/:queue/messages to get messages, i.e. consume or peek from a queue. DELETE /vhosts/:vhost/queues/:queue/messages to delete messages, i.e. purge a queue. ``` A separate new HTTP API v2 allows us to ship only handlers for HTTP over AMQP for RabbitMQ 4.0 and therefore move faster while still keeping the option on the table to re-use the new v2 API for pure HTTP in the future. In contrast, re-using the HTTP API v1 for HTTP over AMQP is possible, but dirty because separate handlers (HTTP over AMQP and pure HTTP) replying differently will be needed for the same v1 endpoints.
2024-02-08 01:26:13 +08:00
- rabbit_amqp_management
Support AMQP 1.0 natively ## What Similar to Native MQTT in #5895, this commits implements Native AMQP 1.0. By "native", we mean do not proxy via AMQP 0.9.1 anymore. ## Why Native AMQP 1.0 comes with the following major benefits: 1. Similar to Native MQTT, this commit provides better throughput, latency, scalability, and resource usage for AMQP 1.0. See https://blog.rabbitmq.com/posts/2023/03/native-mqtt for native MQTT improvements. See further below for some benchmarks. 2. Since AMQP 1.0 is not limited anymore by the AMQP 0.9.1 protocol, this commit allows implementing more AMQP 1.0 features in the future. Some features are already implemented in this commit (see next section). 3. Simpler, better understandable, and more maintainable code. Native AMQP 1.0 as implemented in this commit has the following major benefits compared to AMQP 0.9.1: 4. Memory and disk alarms will only stop accepting incoming TRANSFER frames. New connections can still be created to consume from RabbitMQ to empty queues. 5. Due to 4. no need anymore for separate connections for publishers and consumers as we currently recommended for AMQP 0.9.1. which potentially halves the number of physical TCP connections. 6. When a single connection sends to multiple target queues, a single slow target queue won't block the entire connection. Publisher can still send data quickly to all other target queues. 7. A publisher can request whether it wants publisher confirmation on a per-message basis. In AMQP 0.9.1 publisher confirms are configured per channel only. 8. Consumers can change their "prefetch count" dynamically which isn't possible in our AMQP 0.9.1 implementation. See #10174 9. AMQP 1.0 is an extensible protocol This commit also fixes dozens of bugs present in the AMQP 1.0 plugin in RabbitMQ 3.x - most of which cannot be backported due to the complexity and limitations of the old 3.x implementation. This commit contains breaking changes and is therefore targeted for RabbitMQ 4.0. ## Implementation details 1. Breaking change: With Native AMQP, the behaviour of ``` Convert AMQP 0.9.1 message headers to application properties for an AMQP 1.0 consumer amqp1_0.convert_amqp091_headers_to_app_props = false | true (default false) Convert AMQP 1.0 Application Properties to AMQP 0.9.1 headers amqp1_0.convert_app_props_to_amqp091_headers = false | true (default false) ``` will break because we always convert according to the message container conversions. For example, AMQP 0.9.1 x-headers will go into message-annotations instead of application properties. Also, `false` won’t be respected since we always convert the headers with message containers. 2. Remove rabbit_queue_collector rabbit_queue_collector is responsible for synchronously deleting exclusive queues. Since the AMQP 1.0 plugin never creates exclusive queues, rabbit_queue_collector doesn't need to be started in the first place. This will save 1 Erlang process per AMQP 1.0 connection. 3. 7 processes per connection + 1 process per session in this commit instead of 7 processes per connection + 15 processes per session in 3.x Supervision hierarchy got re-designed. 4. Use 1 writer process per AMQP 1.0 connection AMQP 0.9.1 uses a separate rabbit_writer Erlang process per AMQP 0.9.1 channel. Prior to this commit, AMQP 1.0 used a separate rabbit_amqp1_0_writer process per AMQP 1.0 session. Advantage of single writer proc per session (prior to this commit): * High parallelism for serialising packets if multiple sessions within a connection write heavily at the same time. This commit uses a single writer process per AMQP 1.0 connection that is shared across all AMQP 1.0 sessions. Advantages of single writer proc per connection (this commit): * Lower memory usage with hundreds of thousands of AMQP 1.0 sessions * Less TCP and IP header overhead given that the single writer process can accumulate across all sessions bytes before flushing the socket. In other words, this commit decides that a reader / writer process pair per AMQP 1.0 connection is good enough for bi-directional TRANSFER flows. Having a writer per session is too heavy. We still ensure high throughput by having separate reader, writer, and session processes. 5. Transform rabbit_amqp1_0_writer into gen_server Why: Prior to this commit, when clicking on the AMQP 1.0 writer process in observer, the process crashed. Instead of handling all these debug messages of the sys module, it's better to implement a gen_server. There is no advantage of using a special OTP process over gen_server for the AMQP 1.0 writer. gen_server also provides cleaner format status output. How: Message callbacks return a timeout of 0. After all messages in the inbox are processed, the timeout message is handled by flushing any pending bytes. 6. Remove stats timer from writer AMQP 1.0 connections haven't emitted any stats previously. 7. When there are contiguous queue confirmations in the session process mailbox, batch them. When the confirmations are sent to the publisher, a single DISPOSITION frame is sent for contiguously confirmed delivery IDs. This approach should be good enough. However it's sub optimal in scenarios where contiguous delivery IDs that need confirmations are rare, for example: * There are multiple links in the session with different sender settlement modes and sender publishes across these links interleaved. * sender settlement mode is mixed and sender publishes interleaved settled and unsettled TRANSFERs. 8. Introduce credit API v2 Why: The AMQP 0.9.1 credit extension which is to be removed in 4.0 was poorly designed since basic.credit is a synchronous call into the queue process blocking the entire AMQP 1.0 session process. How: Change the interactions between queue clients and queue server implementations: * Clients only request a credit reply if the FLOW's `echo` field is set * Include all link flow control state held by the queue process into a new credit_reply queue event: * `available` after the queue sends any deliveries * `link-credit` after the queue sends any deliveries * `drain` which allows us to combine the old queue events send_credit_reply and send_drained into a single new queue event credit_reply. * Include the consumer tag into the credit_reply queue event such that the AMQP 1.0 session process can process any credit replies asynchronously. Link flow control state `delivery-count` also moves to the queue processes. The new interactions are hidden behind feature flag credit_api_v2 to allow for rolling upgrades from 3.13 to 4.0. 9. Use serial number arithmetic in quorum queues and session process. 10. Completely bypass the rabbit_limiter module for AMQP 1.0 flow control. The goal is to eventually remove the rabbit_limiter module in 4.0 since AMQP 0.9.1 global QoS will be unsupported in 4.0. This commit lifts the AMQP 1.0 link flow control logic out of rabbit_limiter into rabbit_queue_consumers. 11. Fix credit bug for streams: AMQP 1.0 settlements shouldn't top up link credit, only FLOW frames should top up link credit. 12. Allow sender settle mode unsettled for streams since AMQP 1.0 acknowledgements to streams are no-ops (currently). 13. Fix AMQP 1.0 client bugs Auto renewing credits should not be related to settling TRANSFERs. Remove field link_credit_unsettled as it was wrong and confusing. Prior to this commit auto renewal did not work when the sender uses sender settlement mode settled. 14. Fix AMQP 1.0 client bugs The wrong outdated Link was passed to function auto_flow/2 15. Use osiris chunk iterator Only hold messages of uncompressed sub batches in memory if consumer doesn't have sufficient credits. Compressed sub batches are skipped for non Stream protocol consumers. 16. Fix incoming link flow control Always use confirms between AMQP 1.0 queue clients and queue servers. As already done internally by rabbit_fifo_client and rabbit_stream_queue, use confirms for classic queues as well. 17. Include link handle into correlation when publishing messages to target queues such that session process can correlate confirms from target queues to incoming links. 18. Only grant more credits to publishers if publisher hasn't sufficient credits anymore and there are not too many unconfirmed messages on the link. 19. Completely ignore `block` and `unblock` queue actions and RabbitMQ credit flow between classic queue process and session process. 20. Link flow control is independent between links. A client can refer to a queue or to an exchange with multiple dynamically added target queues. Multiple incoming links can also fan in to the same queue. However the link topology looks like, this commit ensures that each link is only granted more credits if that link isn't overloaded. 21. A connection or a session can send to many different queues. In AMQP 0.9.1, a single slow queue will lead to the entire channel, and then entire connection being blocked. This commit makes sure that a single slow queue from one link won't slow down sending on other links. For example, having link A sending to a local classic queue and link B sending to 5 replica quorum queue, link B will naturally grant credits slower than link A. So, despite the quorum queue being slower in confirming messages, the same AMQP 1.0 connection and session can still pump data very fast into the classic queue. 22. If cluster wide memory or disk alarm occurs. Each session sends a FLOW with incoming-window to 0 to sending client. If sending clients don’t obey, force disconnect the client. If cluster wide memory alarm clears: Each session resumes with a FLOW defaulting to initial incoming-window. 23. All operations apart of publishing TRANSFERS to RabbitMQ can continue during cluster wide alarms, specifically, attaching consumers and consuming, i.e. emptying queues. There is no need for separate AMQP 1.0 connections for publishers and consumers as recommended in our AMQP 0.9.1 implementation. 24. Flow control summary: * If queue becomes bottleneck, that’s solved by slowing down individual sending links (AMQP 1.0 link flow control). * If session becomes bottleneck (more unlikely), that’s solved by AMQP 1.0 session flow control. * If connection becomes bottleneck, it naturally won’t read fast enough from the socket causing TCP backpressure being applied. Nowhere will RabbitMQ internal credit based flow control (i.e. module credit_flow) be used on the incoming AMQP 1.0 message path. 25. Register AMQP sessions Prefer local-only pg over our custom pg_local implementation as pg is a better process group implementation than pg_local. pg_local was identified as bottleneck in tests where many MQTT clients were disconnected at once. 26. Start a local-only pg when Rabbit boots: > A scope can be kept local-only by using a scope name that is unique cluster-wide, e.g. the node name: > pg:start_link(node()). Register AMQP 1.0 connections and sessions with pg. In future we should remove pg_local and instead use the new local-only pg for all registered processes such as AMQP 0.9.1 connections and channels. 27. Requeue messages if link detached Although the spec allows to settle delivery IDs on detached links, RabbitMQ does not respect the 'closed' field of the DETACH frame and therefore handles every DETACH frame as closed. Since the link is closed, we expect every outstanding delivery to be requeued. In addition to consumer cancellation, detaching a link therefore causes in flight deliveries to be requeued. Note that this behaviour is different from merely consumer cancellation in AMQP 0.9.1: "After a consumer is cancelled there will be no future deliveries dispatched to it. Note that there can still be "in flight" deliveries dispatched previously. Cancelling a consumer will neither discard nor requeue them." [https://www.rabbitmq.com/consumers.html#unsubscribing] An AMQP receiver can first drain, and then detach to prevent "in flight" deliveries 28. Init AMQP session with BEGIN frame Similar to how there can't be an MQTT processor without a CONNECT frame, there can't be an AMQP session without a BEGIN frame. This allows having strict dialyzer types for session flow control fields (i.e. not allowing 'undefined'). 29. Move serial_number to AMQP 1.0 common lib such that it can be used by both AMQP 1.0 server and client 30. Fix AMQP client to do serial number arithmetic. 31. AMQP client: Differentiate between delivery-id and transfer-id for better understandability. 32. Fix link flow control in classic queues This commit fixes ``` java -jar target/perf-test.jar -ad false -f persistent -u cq -c 3000 -C 1000000 -y 0 ``` followed by ``` ./omq -x 0 amqp -T /queue/cq -D 1000000 --amqp-consumer-credits 2 ``` Prior to this commit, (and on RabbitMQ 3.x) the consuming would halt after around 8 - 10,000 messages. The bug was that in flight messages from classic queue process to session process were not taken into account when topping up credit to the classic queue process. Fixes #2597 The solution to this bug (and a much cleaner design anyway independent of this bug) is that queues should hold all link flow control state including the delivery-count. Hence, when credit API v2 is used the delivery-count will be held by the classic queue process, quorum queue process, and stream queue client instead of managing the delivery-count in the session. 33. The double level crediting between (a) session process and rabbit_fifo_client, and (b) rabbit_fifo_client and rabbit_fifo was removed. Therefore, instead of managing 3 separate delivery-counts (i. session, ii. rabbit_fifo_client, iii. rabbit_fifo), only 1 delivery-count is used in rabbit_fifo. This is a big simplification. 34. This commit fixes quorum queues without bumping the machine version nor introducing new rabbit_fifo commands. Whether credit API v2 is used is solely determined at link attachment time depending on whether feature flag credit_api_v2 is enabled. Even when that feature flag will be enabled later on, this link will keep using credit API v1 until detached (or the node is shut down). Eventually, after feature flag credit_api_v2 has been enabled and a subsequent rolling upgrade, all links will use credit API v2. This approach is safe and simple. The 2 alternatives to move delivery-count from the session process to the queue processes would have been: i. Explicit feature flag credit_api_v2 migration function * Can use a gen_server:call and only finish migration once all delivery-counts were migrated. Cons: * Extra new message format just for migration is required. * Risky as migration will fail if a target queue doesn’t reply. ii. Session always includes DeliveryCountSnd when crediting to the queue: Cons: * 2 delivery counts will be hold simultaneously in session proc and queue proc; could be solved by deleting the session proc’s delivery-count for credit-reply * What happens if the receiver doesn’t provide credit for a very long time? Is that a problem? 35. Support stream filtering in AMQP 1.0 (by @acogoluegnes) Use the x-stream-filter-value message annotation to carry the filter value in a published message. Use the rabbitmq:stream-filter and rabbitmq:stream-match-unfiltered filters when creating a receiver that wants to filter out messages from a stream. 36. Remove credit extension from AMQP 0.9.1 client 37. Support maintenance mode closing AMQP 1.0 connections. 38. Remove AMQP 0.9.1 client dependency from AMQP 1.0 implementation. 39. Move AMQP 1.0 plugin to the core. AMQP 1.0 is enabled by default. The old rabbitmq_amqp1_0 plugin will be kept as a no-op plugin to prevent deployment tools from failing that execute: ``` rabbitmq-plugins enable rabbitmq_amqp1_0 rabbitmq-plugins disable rabbitmq_amqp1_0 ``` 40. Breaking change: Remove CLI command `rabbitmqctl list_amqp10_connections`. Instead, list both AMQP 0.9.1 and AMQP 1.0 connections in `list_connections`: ``` rabbitmqctl list_connections protocol Listing connections ... protocol {1, 0} {0,9,1} ``` ## Benchmarks ### Throughput & Latency Setup: * Single node Ubuntu 22.04 * Erlang 26.1.1 Start RabbitMQ: ``` make run-broker PLUGINS="rabbitmq_management rabbitmq_amqp1_0" FULL=1 RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+S 3" ``` Predeclare durable classic queue cq1, durable quorum queue qq1, durable stream queue sq1. Start client: https://github.com/ssorj/quiver https://hub.docker.com/r/ssorj/quiver/tags (digest 453a2aceda64) ``` docker run -it --rm --add-host host.docker.internal:host-gateway ssorj/quiver:latest bash-5.1# quiver --version quiver 0.4.0-SNAPSHOT ``` 1. Classic queue ``` quiver //host.docker.internal//amq/queue/cq1 --durable --count 1m --duration 10m --body-size 12 --credit 1000 ``` This commit: ``` Count ............................................. 1,000,000 messages Duration ............................................... 73.8 seconds Sender rate .......................................... 13,548 messages/s Receiver rate ........................................ 13,547 messages/s End-to-end rate ...................................... 13,547 messages/s Latencies by percentile: 0% ........ 0 ms 90.00% ........ 9 ms 25% ........ 2 ms 99.00% ....... 14 ms 50% ........ 4 ms 99.90% ....... 17 ms 100% ....... 26 ms 99.99% ....... 24 ms ``` RabbitMQ 3.x (main branch as of 30 January 2024): ``` ---------------------- Sender ----------------------- --------------------- Receiver ---------------------- -------- Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Lat [ms] ----------------------------------------------------- ----------------------------------------------------- -------- 2.1 130,814 65,342 6 73.6 2.1 3,217 1,607 0 8.0 511 4.1 163,580 16,367 2 74.1 4.1 3,217 0 0 8.0 0 6.1 229,114 32,767 3 74.1 6.1 3,217 0 0 8.0 0 8.1 261,880 16,367 2 74.1 8.1 67,874 32,296 8 8.2 7,662 10.1 294,646 16,367 2 74.1 10.1 67,874 0 0 8.2 0 12.1 360,180 32,734 3 74.1 12.1 67,874 0 0 8.2 0 14.1 392,946 16,367 3 74.1 14.1 68,604 365 0 8.2 12,147 16.1 458,480 32,734 3 74.1 16.1 68,604 0 0 8.2 0 18.1 491,246 16,367 2 74.1 18.1 68,604 0 0 8.2 0 20.1 556,780 32,767 4 74.1 20.1 68,604 0 0 8.2 0 22.1 589,546 16,375 2 74.1 22.1 68,604 0 0 8.2 0 receiver timed out 24.1 622,312 16,367 2 74.1 24.1 68,604 0 0 8.2 0 quiver: error: PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/cq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-otujr23y' returned non-zero exit status 1. Traceback (most recent call last): File "/usr/local/lib/quiver/python/quiver/pair.py", line 144, in run _plano.wait(receiver, check=True) File "/usr/local/lib/quiver/python/plano/main.py", line 1243, in wait raise PlanoProcessError(proc) plano.main.PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/cq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-otujr23y' returned non-zero exit status 1. ``` 2. Quorum queue: ``` quiver //host.docker.internal//amq/queue/qq1 --durable --count 1m --duration 10m --body-size 12 --credit 1000 ``` This commit: ``` Count ............................................. 1,000,000 messages Duration .............................................. 101.4 seconds Sender rate ........................................... 9,867 messages/s Receiver rate ......................................... 9,868 messages/s End-to-end rate ....................................... 9,865 messages/s Latencies by percentile: 0% ....... 11 ms 90.00% ....... 23 ms 25% ....... 15 ms 99.00% ....... 28 ms 50% ....... 18 ms 99.90% ....... 33 ms 100% ....... 49 ms 99.99% ....... 47 ms ``` RabbitMQ 3.x: ``` ---------------------- Sender ----------------------- --------------------- Receiver ---------------------- -------- Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Lat [ms] ----------------------------------------------------- ----------------------------------------------------- -------- 2.1 130,814 65,342 9 69.9 2.1 18,430 9,206 5 7.6 1,221 4.1 163,580 16,375 5 70.2 4.1 18,867 218 0 7.6 2,168 6.1 229,114 32,767 6 70.2 6.1 18,867 0 0 7.6 0 8.1 294,648 32,734 7 70.2 8.1 18,867 0 0 7.6 0 10.1 360,182 32,734 6 70.2 10.1 18,867 0 0 7.6 0 12.1 425,716 32,767 6 70.2 12.1 18,867 0 0 7.6 0 receiver timed out 14.1 458,482 16,367 5 70.2 14.1 18,867 0 0 7.6 0 quiver: error: PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/qq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-b1gcup43' returned non-zero exit status 1. Traceback (most recent call last): File "/usr/local/lib/quiver/python/quiver/pair.py", line 144, in run _plano.wait(receiver, check=True) File "/usr/local/lib/quiver/python/plano/main.py", line 1243, in wait raise PlanoProcessError(proc) plano.main.PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/qq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-b1gcup43' returned non-zero exit status 1. ``` 3. Stream: ``` quiver-arrow send //host.docker.internal//amq/queue/sq1 --durable --count 1m -d 10m --summary --verbose ``` This commit: ``` Count ............................................. 1,000,000 messages Duration ................................................ 8.7 seconds Message rate ........................................ 115,154 messages/s ``` RabbitMQ 3.x: ``` Count ............................................. 1,000,000 messages Duration ............................................... 21.2 seconds Message rate ......................................... 47,232 messages/s ``` ### Memory usage Start RabbitMQ: ``` ERL_MAX_PORTS=3000000 RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+P 3000000 +S 6" make run-broker PLUGINS="rabbitmq_amqp1_0" FULL=1 RABBITMQ_CONFIG_FILE="rabbitmq.conf" ``` ``` /bin/cat rabbitmq.conf tcp_listen_options.sndbuf = 2048 tcp_listen_options.recbuf = 2048 vm_memory_high_watermark.relative = 0.95 vm_memory_high_watermark_paging_ratio = 0.95 loopback_users = none ``` Create 50k connections with 2 sessions per connection, i.e. 100k session in total: ```go package main import ( "context" "log" "time" "github.com/Azure/go-amqp" ) func main() { for i := 0; i < 50000; i++ { conn, err := amqp.Dial(context.TODO(), "amqp://nuc", &amqp.ConnOptions{SASLType: amqp.SASLTypeAnonymous()}) if err != nil { log.Fatal("dialing AMQP server:", err) } _, err = conn.NewSession(context.TODO(), nil) if err != nil { log.Fatal("creating AMQP session:", err) } _, err = conn.NewSession(context.TODO(), nil) if err != nil { log.Fatal("creating AMQP session:", err) } } log.Println("opened all connections") time.Sleep(5 * time.Hour) } ``` This commit: ``` erlang:memory(). [{total,4586376480}, {processes,4025898504}, {processes_used,4025871040}, {system,560477976}, {atom,1048841}, {atom_used,1042841}, {binary,233228608}, {code,21449982}, {ets,108560464}] erlang:system_info(process_count). 450289 ``` 7 procs per connection + 1 proc per session. (7 + 2*1) * 50,000 = 450,000 procs RabbitMQ 3.x: ``` erlang:memory(). [{total,15168232704}, {processes,14044779256}, {processes_used,14044755120}, {system,1123453448}, {atom,1057033}, {atom_used,1052587}, {binary,236381264}, {code,21790238}, {ets,391423744}] erlang:system_info(process_count). 1850309 ``` 7 procs per connection + 15 per session (7 + 2*15) * 50,000 = 1,850,000 procs 50k connections + 100k session require with this commit: 4.5 GB in RabbitMQ 3.x: 15 GB ## Future work 1. More efficient parser and serializer 2. TODO in mc_amqp: Do not store the parsed message on disk. 3. Implement both AMQP HTTP extension and AMQP management extension to allow AMQP clients to create RabbitMQ objects (queues, exchanges, ...).
2023-07-21 18:29:07 +08:00
- rabbit_amqp_reader
- rabbit_amqp_session
- rabbit_amqp_session_sup
- rabbit_amqp_util
- rabbit_amqp_writer
- rabbit_amqqueue
- rabbit_amqqueue_control
- rabbit_amqqueue_process
- rabbit_amqqueue_sup
- rabbit_amqqueue_sup_sup
- rabbit_auth_backend_internal
- rabbit_auth_mechanism_amqplain
Add SASL mechanism ANONYMOUS ## 1. Introduce new SASL mechanism ANONYMOUS ### What? Introduce a new `rabbit_auth_mechanism` implementation for SASL mechanism ANONYMOUS called `rabbit_auth_mechanism_anonymous`. ### Why? As described in AMQP section 5.3.3.1, ANONYMOUS should be used when the client doesn't need to authenticate. Introducing a new `rabbit_auth_mechanism` consolidates and simplifies how anonymous logins work across all RabbitMQ protocols that support SASL. This commit therefore allows AMQP 0.9.1, AMQP 1.0, stream clients to connect out of the box to RabbitMQ without providing any username or password. Today's AMQP 0.9.1 and stream protocol client libs hard code RabbitMQ default credentials `guest:guest` for example done in: * https://github.com/rabbitmq/rabbitmq-java-client/blob/0215e85643a9ae0800822869be0200024e2ab569/src/main/java/com/rabbitmq/client/ConnectionFactory.java#L58-L61 * https://github.com/rabbitmq/amqp091-go/blob/ddb7a2f0685689063e6d709b8e417dbf9d09469c/uri.go#L31-L32 Hard coding RabbitMQ specific default credentials in dozens of different client libraries is an anti-pattern in my opinion. Furthermore, there are various AMQP 1.0 and MQTT client libraries which we do not control or maintain and which still should work out of the box when a user is getting started with RabbitMQ (that is without providing `guest:guest` credentials). ### How? The old RabbitMQ 3.13 AMQP 1.0 plugin `default_user` [configuration](https://github.com/rabbitmq/rabbitmq-server/blob/146b4862d8e570b344c99c37d91246760e218b18/deps/rabbitmq_amqp1_0/Makefile#L6) is replaced with the following two new `rabbit` configurations: ``` {anonymous_login_user, <<"guest">>}, {anonymous_login_pass, <<"guest">>}, ``` We call it `anonymous_login_user` because this user will be used for anonymous logins. The subsequent commit uses the same setting for anonymous logins in MQTT. Hence, this user is orthogonal to the protocol used when the client connects. Setting `anonymous_login_pass` could have been left out. This commit decides to include it because our documentation has so far recommended: > It is highly recommended to pre-configure a new user with a generated username and password or delete the guest user > or at least change its password to reasonably secure generated value that won't be known to the public. By having the new module `rabbit_auth_mechanism_anonymous` internally authenticate with `anonymous_login_pass` instead of blindly allowing access without any password, we protect operators that relied on the sentence: > or at least change its password to reasonably secure generated value that won't be known to the public To ease the getting started experience, since RabbitMQ already deploys a guest user with full access to the default virtual host `/`, this commit also allows SASL mechanism ANONYMOUS in `rabbit` setting `auth_mechanisms`. In production, operators should disable SASL mechanism ANONYMOUS by setting `anonymous_login_user` to `none` (or by removing ANONYMOUS from the `auth_mechanisms` setting. This will be documented separately. Even if operators forget or don't read the docs, this new ANONYMOUS mechanism won't do any harm because it relies on the default user name `guest` and password `guest`, which is recommended against in production, and who by default can only connect from the local host. ## 2. Require SASL security layer in AMQP 1.0 ### What? An AMQP 1.0 client must use the SASL security layer. ### Why? This is in line with the mandatory usage of SASL in AMQP 0.9.1 and RabbitMQ stream protocol. Since (presumably) any AMQP 1.0 client knows how to authenticate with a username and password using SASL mechanism PLAIN, any AMQP 1.0 client also (presumably) implements the trivial SASL mechanism ANONYMOUS. Skipping SASL is not recommended in production anyway. By requiring SASL, configuration for operators becomes easier. Following the principle of least surprise, when an an operator configures `auth_mechanisms` to exclude `ANONYMOUS`, anonymous logins will be prohibited in SASL and also by disallowing skipping the SASL layer. ### How? This commit implements AMQP 1.0 figure 2.13. A follow-up commit needs to be pushed to `v3.13.x` which will use SASL mechanism `anon` instead of `none` in the Erlang AMQP 1.0 client such that AMQP 1.0 shovels running on 3.13 can connect to 4.0 RabbitMQ nodes.
2024-08-14 18:19:17 +08:00
- rabbit_auth_mechanism_anonymous
- rabbit_auth_mechanism_cr_demo
- rabbit_auth_mechanism_plain
- rabbit_autoheal
- rabbit_backing_queue
- rabbit_basic
- rabbit_binding
- rabbit_boot_steps
- rabbit_channel
- rabbit_channel_interceptor
- rabbit_channel_sup
- rabbit_channel_sup_sup
- rabbit_channel_tracking
- rabbit_channel_tracking_handler
- rabbit_classic_queue
- rabbit_classic_queue_index_v2
- rabbit_classic_queue_store_v2
- rabbit_client_sup
- rabbit_config
- rabbit_confirms
- rabbit_connection_helper_sup
- rabbit_connection_sup
- rabbit_connection_tracking
- rabbit_connection_tracking_handler
- rabbit_control_pbe
- rabbit_core_ff
- rabbit_core_metrics_gc
- rabbit_credential_validation
- rabbit_credential_validator
- rabbit_credential_validator_accept_everything
- rabbit_credential_validator_min_password_length
- rabbit_credential_validator_password_regexp
- rabbit_cuttlefish
- rabbit_db
- rabbit_db_binding
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
- rabbit_db_binding_m2k_converter
- rabbit_db_cluster
- rabbit_db_exchange
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
- rabbit_db_exchange_m2k_converter
- rabbit_db_m2k_converter
- rabbit_db_maintenance
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
- rabbit_db_maintenance_m2k_converter
- rabbit_db_msup
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
- rabbit_db_msup_m2k_converter
- rabbit_db_policy
- rabbit_db_queue
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
- rabbit_db_queue_m2k_converter
- rabbit_db_rtparams
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
- rabbit_db_rtparams_m2k_converter
- rabbit_db_topic_exchange
- rabbit_db_user
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
- rabbit_db_user_m2k_converter
- rabbit_db_vhost
- rabbit_db_vhost_defaults
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
- rabbit_db_vhost_m2k_converter
- rabbit_dead_letter
- rabbit_definitions
- rabbit_definitions_hashing
- rabbit_definitions_import_https
- rabbit_definitions_import_local_filesystem
2023-12-13 12:02:34 +08:00
- rabbit_depr_ff_extra
Deprecated features: New module to manage deprecated features (!) This introduces a way to declare deprecated features in the code, not only in our communication. The new module allows to disallow the use of a deprecated feature and/or warn the user when he relies on such a feature. [Why] Currently, we only tell people about deprecated features through blog posts and the mailing-list. This might be insufficiant for our users that a feature they use will be removed in a future version: * They may not read our blog or mailing-list * They may not understand that they use such a deprecated feature * They might wait for the big removal before they plan testing * They might not take it seriously enough The idea behind this patch is to increase the chance that users notice that they are using something which is about to be dropped from RabbitMQ. Anopther benefit is that they should be able to test how RabbitMQ will behave in the future before the actual removal. This should allow them to test and plan changes. [How] When a feature is deprecated in other large projects (such as FreeBSD where I took the idea from), it goes through a lifecycle: 1. The feature is still available, but users get a warning somehow when they use it. They can disable it to test. 2. The feature is still available, but disabled out-of-the-box. Users can re-enable it (and get a warning). 3. The feature is disconnected from the build. Therefore, the code behind it is still there, but users have to recompile the thing to be able to use it. 4. The feature is removed from the source code. Users have to adapt or they can't upgrade anymore. The solution in this patch offers the same lifecycle. A deprecated feature will be in one of these deprecation phases: 1. `permitted_by_default`: The feature is available. Users get a warning if they use it. They can disable it from the configuration. 2. `denied_by_default`: The feature is available but disabled by default. Users get an error if they use it and RabbitMQ behaves like the feature is removed. They can re-enable is from the configuration and get a warning. 3. `disconnected`: The feature is present in the source code, but is disabled and can't be re-enabled without recompiling RabbitMQ. Users get the same behavior as if the code was removed. 4. `removed`: The feature's code is gone. The whole thing is based on the feature flags subsystem, but it has the following differences with other feature flags: * The semantic is reversed: the feature flag behind a deprecated feature is disabled when the deprecated feature is permitted, or enabled when the deprecated feature is denied. * The feature flag behind a deprecated feature is enabled out-of-the-box (meaning the deprecated feature is denied): * if the deprecation phase is `permitted_by_default` and the configuration denies the deprecated feature * if the deprecation phase is `denied_by_default` and the configuration doesn't permit the deprecated feature * if the deprecation phase is `disconnected` or `removed` * Feature flags behind deprecated feature don't appear in feature flags listings. Otherwise, deprecated features' feature flags are managed like other feature flags, in particular inside clusters. To declare a deprecated feature: -rabbit_deprecated_feature( {my_deprecated_feature, #{deprecation_phase => permitted_by_default, msgs => #{when_permitted => "This feature will be removed in RabbitMQ X.0"}, }}). Then, to check the state of a deprecated feature in the code: case rabbit_deprecated_features:is_permitted(my_deprecated_feature) of true -> %% The deprecated feature is still permitted. ok; false -> %% The deprecated feature is gone or should be considered %% unavailable. error end. Warnings and errors are logged automatically. A message is generated automatically, but it is possible to define a message in the deprecated feature flag declaration like in the example above. Here is an example of a logged warning that was generated automatically: Feature `my_deprecated_feature` is deprecated. By default, this feature can still be used for now. Its use will not be permitted by default in a future minor RabbitMQ version and the feature will be removed from a future major RabbitMQ version; actual versions to be determined. To continue using this feature when it is not permitted by default, set the following parameter in your configuration: "deprecated_features.permit.my_deprecated_feature = true" To test RabbitMQ as if the feature was removed, set this in your configuration: "deprecated_features.permit.my_deprecated_feature = false" To override the default state of `permitted_by_default` and `denied_by_default` deprecation phases, users can set the following configuration: # In rabbitmq.conf: deprecated_features.permit.my_deprecated_feature = true # or false The actual behavior protected by a deprecated feature check is out of scope for this subsystem. It is the repsonsibility of each deprecated feature code to determine what to do when the deprecated feature is denied. V1: Deprecated feature states are initially computed during the initialization of the registry, based on their deprecation phase and possibly the configuration. They don't go through the `enable/1` code at all. V2: Manage deprecated feature states as any other non-required feature flags. This allows to execute an `is_feature_used()` callback to determine if a deprecated feature can be denied. This also allows to prevent the RabbitMQ node from starting if it continues to use a deprecated feature. V3: Manage deprecated feature states from the registry initialization again. This is required because we need to know very early if some of them are denied, so that an upgrade to a version of RabbitMQ where a deprecated feature is disconnected or removed can be performed. To still prevent the start of a RabbitMQ node when a denied deprecated feature is actively used, we run the `is_feature_used()` callback of all denied deprecated features as part of the `sync_cluster()` task. This task is executed as part of a feature flag refresh executed when RabbitMQ starts or when plugins are enabled. So even though a deprecated feature is marked as denied in the registry early in the boot process, we will still abort the start of a RabbitMQ node if the feature is used. V4: Support context-dependent warnings. It is now possible to set a specific message when deprecated feature is permitted, when it is denied and when it is removed. Generic per-context messages are still generated. V5: Improve default warning messages, thanks to @pstack2021. V6: Rename the configuration variable from `permit_deprecated_features.*` to `deprecated_features.permit.*`. As @michaelklishin said, we tend to use shorter top-level names.
2023-02-23 00:26:52 +08:00
- rabbit_deprecated_features
- rabbit_diagnostics
- rabbit_direct
- rabbit_direct_reply_to
- rabbit_disk_monitor
- rabbit_epmd_monitor
- rabbit_event_consumer
- rabbit_exchange
- rabbit_exchange_decorator
- rabbit_exchange_parameters
- rabbit_exchange_type
- rabbit_exchange_type_direct
- rabbit_exchange_type_fanout
- rabbit_exchange_type_headers
- rabbit_exchange_type_invalid
- rabbit_exchange_type_local_random
- rabbit_exchange_type_topic
- rabbit_feature_flags
- rabbit_ff_controller
- rabbit_ff_extra
- rabbit_ff_registry
- rabbit_ff_registry_factory
2023-05-23 12:02:30 +08:00
- rabbit_ff_registry_wrapper
- rabbit_fhc_helpers
- rabbit_fifo
- rabbit_fifo_client
- rabbit_fifo_dlx
- rabbit_fifo_dlx_client
- rabbit_fifo_dlx_sup
- rabbit_fifo_dlx_worker
- rabbit_fifo_index
Quorum queues v4 (#10637) This commit contains the following new quorum queue features: * Fair share high/low priorities * SAC consumers honour consumer priorities * Credited consumer refactoring to meet AMQP requirements. * Use checkpoints feature to reduce memory use for queues with long backlogs * Consumer cancel option that immediately removes consumer and returns all pending messages. * More compact commands of the most common commands such as enqueue, settle and credit * Correctly track the delivery-count to be compatible with the AMQP spec * Support the "modified" AMQP 1.0 outcome better. Commits: * Quorum queues v4 scaffolding. Create the new version but not including any changes yet. QQ: force delete followers after leader has terminated. Also try a longer sleep for mqtt_shared_SUITE so that the delete operation stands a chance to time out and move on to the forced deletion stage. In some mixed machine version scenarios some followers will never apply the poison pill command so we may as well force delete them just in case. QQ: skip test in amqp_client that cannot pass with mixed machine versions QQ: remove dead code Code relating to prior machine versions and state conversions. rabbit_fifo_prop_SUITE fixes * QQ: add v4 ff and new more compact enqueue command. Also update rabbit_fifo_* suites to test more relevant code versions where applicable. QQ: always use the updated credit mode format QQv4: use more compact consumer reference in settle, credit, return This introudces a new type: consumer_key() which is either the consumer_id or the raft index the checkout was processed at. If the consumer is using one of the updated credit spec formats rabbit_fifo will use the raft index as the primary key for the consumer such that the rabbit fifo client can then use the more space efficient integer index instead of the full consumer id in subsequent commands. There is compatibility code to still accept the consumer id in settle, return, discard and credit commands but this is slighlyt slower and of course less space efficient. The old form will be used in cases where the fifo client may have already remove the local consumer state (as happens after a cancel). Lots of test refactorings of the rabbit_fifo_SUITE to begin to use the new forms. * More test refactoring and new API fixes rabbit_fifo_prop_SUITE refactoring and other fixes. * First pass SAC consumer priority implementation. Single active consumers will be activated if they have a higher priority than the currently active consumer. if the currently active consumer has pending messages, no further messages will be assigned to the consumer and the activation of the new consumer will happen once all pending messages are settled. This is to ensure processing order. Consumers with the same priority will internally be ordered to favour those with credit then those that attached first. QQ: add SAC consumer priority integration tests QQ: add check for ff in tests * QQ: add new consumer cancel option: 'remove' This option immediately removes and returns all messages for a consumer instead of the softer 'cancel' option which keeps the consumer around until all pending messages have been either settled or returned. This involves a change to the rabbit_queue_type:cancel/5 API to rabbit_queue_type:cancel/3. * QQ: capture checked out time for each consumer message. This will form the basis for queue initiated consumer timeouts. * QQ: Refactor to use the new ra_machine:handle_aux/5 API Instead of the old ra_machine:handle_aux/6 callback. * QQ hi/lo priority queue * QQ: Avoid using mc:size/1 inside rabbit_fifo As we dont want to depend on external functions for things that may change the state of the queue. * QQ bug fix: Maintain order when returning multiple Prior to this commit, quorum queues requeued messages in an undefined order, which is wrong. This commit fixes this bug and requeues messages always in the order as nacked / rejected / released by the client. We ensure that order of requeues is deterministic from the client's point of view and doesn't depend on whether the quorum queue soft limit was exceeded temporarily. So, even when rabbit_fifo_client batches requeues, the order as nacked by the client is still maintained. * Simplify * Add rabbit_quorum_queue:file_handle* functions back. For backwards compat. * dialyzer fix * dynamic_qq_SUITE: avoid mixed versions failure. * QQ: track number of requeues for message. To be able to calculate the correct value for the AMQP delivery_count header we need to be able to distinguish between messages that were "released" or returned in QQ speak and those that were returned due to errors such as channel termination. This commit implement such tracking as well as the calculation of a new mc annotations `delivery_count` that AMQP makes use of to set the header value accordingly. * Use QQ consumer removal when AMQP client detaches This enables us to unskip some AMQP tests. * Use AMQP address v2 in fsharp-tests * QQ: track number of requeues for message. To be able to calculate the correct value for the AMQP delivery_count header we need to be able to distinguish between messages that were "released" or returned in QQ speak and those that were returned due to errors such as channel termination. This commit implement such tracking as well as the calculation of a new mc annotations `delivery_count` that AMQP makes use of to set the header value accordingly. * rabbit_fifo: Use Ra checkpoints * quorum queues: Use a custom interval for checkpoints * rabbit_fifo_SUITE: List actual effects in ?ASSERT_EFF failure * QQ: Checkpoints modifications * fixes * QQ: emit release cursors on tick for followers and leaders else followers could end up holding on to segments a bit longer after traffic stops. * Support draining a QQ SAC waiting consumer By issuing drain=true, the client says "either send a transfer or a flow frame". Since there are no messages to send to an inactive consumer, the sending queue should advance the delivery-count consuming all link-credit and send a credit_reply with drain=true to the session proc which causes the session proc to send a flow frame to the client. * Extract applying #credit{} cmd into 2 functions This commit is only refactoring and doesn't change any behaviour. * Fix default priority level Prior to this commit, when a message didn't have a priority level set, it got enqueued as high prio. This is wrong because the default priority is 4 and "for example, if 2 distinct priorities are implemented, then levels 0 to 4 are equivalent, and levels 5 to 9 are equivalent and levels 4 and 5 are distinct." Hence, by default a message without priority set, must be enqueued as low prio. * bazel run gazelle * Avoid deprecated time unit * Fix aux_test * Delete dead code * Fix rabbit_fifo_q:get_lowest_index/1 * Delete unused normalize functions * Generate less garbage * Add integration test for QQ SAC with consumer priority * Improve readability * Change modified outcome behaviour With the new quorum queue v4 improvements where a requeue counter was added in addition to the quorum queue delivery counter, the following sentence from https://github.com/rabbitmq/rabbitmq-server/pull/6292#issue-1431275848 doesn't apply anymore: > Also the case where delivery_failed=false|undefined requires the release of the > message without incrementing the delivery_count. Again this is not something > that our queues are able to do so again we have to reject without requeue. Therefore, we simplify the modified outcome behaviour: RabbitMQ will from now on only discard the message if the modified's undeliverable-here field is true. * Introduce single feature flag rabbitmq_4.0.0 ## What? Merge all feature flags introduced in RabbitMQ 4.0.0 into a single feature flag called rabbitmq_4.0.0. ## Why? 1. This fixes the crash in https://github.com/rabbitmq/rabbitmq-server/pull/10637#discussion_r1681002352 2. It's better user experience. * QQ: expose priority metrics in UI * Enable skipped test after rebasing onto main * QQ: add new command "modify" to better handle AMQP modified outcomes. This new command can be used to annotate returned or rejected messages. This commit also retains the delivery-count across dead letter boundaries such that the AMQP header delivery-count field can now include _all_ failed deliver attempts since the message was originally received. Internally the quorum queue has moved it's delivery_count header to only track the AMQP protocol delivery attempts and now introduces a new acquired_count to track all message acquisitions by consumers. * Type tweaks and naming * Add test for modified outcome with classic queue * Add test routing on message-annotations in modified outcome * Skip tests in mixed version tests Skip tests in mixed version tests because feature flag rabbitmq_4.0.0 is needed for the new #modify{} Ra command being sent to quorum queues. --------- Co-authored-by: David Ansari <david.ansari@gmx.de> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2024-08-08 15:48:27 +08:00
- rabbit_fifo_q
- rabbit_fifo_v0
- rabbit_fifo_v1
Quorum queues v4 (#10637) This commit contains the following new quorum queue features: * Fair share high/low priorities * SAC consumers honour consumer priorities * Credited consumer refactoring to meet AMQP requirements. * Use checkpoints feature to reduce memory use for queues with long backlogs * Consumer cancel option that immediately removes consumer and returns all pending messages. * More compact commands of the most common commands such as enqueue, settle and credit * Correctly track the delivery-count to be compatible with the AMQP spec * Support the "modified" AMQP 1.0 outcome better. Commits: * Quorum queues v4 scaffolding. Create the new version but not including any changes yet. QQ: force delete followers after leader has terminated. Also try a longer sleep for mqtt_shared_SUITE so that the delete operation stands a chance to time out and move on to the forced deletion stage. In some mixed machine version scenarios some followers will never apply the poison pill command so we may as well force delete them just in case. QQ: skip test in amqp_client that cannot pass with mixed machine versions QQ: remove dead code Code relating to prior machine versions and state conversions. rabbit_fifo_prop_SUITE fixes * QQ: add v4 ff and new more compact enqueue command. Also update rabbit_fifo_* suites to test more relevant code versions where applicable. QQ: always use the updated credit mode format QQv4: use more compact consumer reference in settle, credit, return This introudces a new type: consumer_key() which is either the consumer_id or the raft index the checkout was processed at. If the consumer is using one of the updated credit spec formats rabbit_fifo will use the raft index as the primary key for the consumer such that the rabbit fifo client can then use the more space efficient integer index instead of the full consumer id in subsequent commands. There is compatibility code to still accept the consumer id in settle, return, discard and credit commands but this is slighlyt slower and of course less space efficient. The old form will be used in cases where the fifo client may have already remove the local consumer state (as happens after a cancel). Lots of test refactorings of the rabbit_fifo_SUITE to begin to use the new forms. * More test refactoring and new API fixes rabbit_fifo_prop_SUITE refactoring and other fixes. * First pass SAC consumer priority implementation. Single active consumers will be activated if they have a higher priority than the currently active consumer. if the currently active consumer has pending messages, no further messages will be assigned to the consumer and the activation of the new consumer will happen once all pending messages are settled. This is to ensure processing order. Consumers with the same priority will internally be ordered to favour those with credit then those that attached first. QQ: add SAC consumer priority integration tests QQ: add check for ff in tests * QQ: add new consumer cancel option: 'remove' This option immediately removes and returns all messages for a consumer instead of the softer 'cancel' option which keeps the consumer around until all pending messages have been either settled or returned. This involves a change to the rabbit_queue_type:cancel/5 API to rabbit_queue_type:cancel/3. * QQ: capture checked out time for each consumer message. This will form the basis for queue initiated consumer timeouts. * QQ: Refactor to use the new ra_machine:handle_aux/5 API Instead of the old ra_machine:handle_aux/6 callback. * QQ hi/lo priority queue * QQ: Avoid using mc:size/1 inside rabbit_fifo As we dont want to depend on external functions for things that may change the state of the queue. * QQ bug fix: Maintain order when returning multiple Prior to this commit, quorum queues requeued messages in an undefined order, which is wrong. This commit fixes this bug and requeues messages always in the order as nacked / rejected / released by the client. We ensure that order of requeues is deterministic from the client's point of view and doesn't depend on whether the quorum queue soft limit was exceeded temporarily. So, even when rabbit_fifo_client batches requeues, the order as nacked by the client is still maintained. * Simplify * Add rabbit_quorum_queue:file_handle* functions back. For backwards compat. * dialyzer fix * dynamic_qq_SUITE: avoid mixed versions failure. * QQ: track number of requeues for message. To be able to calculate the correct value for the AMQP delivery_count header we need to be able to distinguish between messages that were "released" or returned in QQ speak and those that were returned due to errors such as channel termination. This commit implement such tracking as well as the calculation of a new mc annotations `delivery_count` that AMQP makes use of to set the header value accordingly. * Use QQ consumer removal when AMQP client detaches This enables us to unskip some AMQP tests. * Use AMQP address v2 in fsharp-tests * QQ: track number of requeues for message. To be able to calculate the correct value for the AMQP delivery_count header we need to be able to distinguish between messages that were "released" or returned in QQ speak and those that were returned due to errors such as channel termination. This commit implement such tracking as well as the calculation of a new mc annotations `delivery_count` that AMQP makes use of to set the header value accordingly. * rabbit_fifo: Use Ra checkpoints * quorum queues: Use a custom interval for checkpoints * rabbit_fifo_SUITE: List actual effects in ?ASSERT_EFF failure * QQ: Checkpoints modifications * fixes * QQ: emit release cursors on tick for followers and leaders else followers could end up holding on to segments a bit longer after traffic stops. * Support draining a QQ SAC waiting consumer By issuing drain=true, the client says "either send a transfer or a flow frame". Since there are no messages to send to an inactive consumer, the sending queue should advance the delivery-count consuming all link-credit and send a credit_reply with drain=true to the session proc which causes the session proc to send a flow frame to the client. * Extract applying #credit{} cmd into 2 functions This commit is only refactoring and doesn't change any behaviour. * Fix default priority level Prior to this commit, when a message didn't have a priority level set, it got enqueued as high prio. This is wrong because the default priority is 4 and "for example, if 2 distinct priorities are implemented, then levels 0 to 4 are equivalent, and levels 5 to 9 are equivalent and levels 4 and 5 are distinct." Hence, by default a message without priority set, must be enqueued as low prio. * bazel run gazelle * Avoid deprecated time unit * Fix aux_test * Delete dead code * Fix rabbit_fifo_q:get_lowest_index/1 * Delete unused normalize functions * Generate less garbage * Add integration test for QQ SAC with consumer priority * Improve readability * Change modified outcome behaviour With the new quorum queue v4 improvements where a requeue counter was added in addition to the quorum queue delivery counter, the following sentence from https://github.com/rabbitmq/rabbitmq-server/pull/6292#issue-1431275848 doesn't apply anymore: > Also the case where delivery_failed=false|undefined requires the release of the > message without incrementing the delivery_count. Again this is not something > that our queues are able to do so again we have to reject without requeue. Therefore, we simplify the modified outcome behaviour: RabbitMQ will from now on only discard the message if the modified's undeliverable-here field is true. * Introduce single feature flag rabbitmq_4.0.0 ## What? Merge all feature flags introduced in RabbitMQ 4.0.0 into a single feature flag called rabbitmq_4.0.0. ## Why? 1. This fixes the crash in https://github.com/rabbitmq/rabbitmq-server/pull/10637#discussion_r1681002352 2. It's better user experience. * QQ: expose priority metrics in UI * Enable skipped test after rebasing onto main * QQ: add new command "modify" to better handle AMQP modified outcomes. This new command can be used to annotate returned or rejected messages. This commit also retains the delivery-count across dead letter boundaries such that the AMQP header delivery-count field can now include _all_ failed deliver attempts since the message was originally received. Internally the quorum queue has moved it's delivery_count header to only track the AMQP protocol delivery attempts and now introduces a new acquired_count to track all message acquisitions by consumers. * Type tweaks and naming * Add test for modified outcome with classic queue * Add test routing on message-annotations in modified outcome * Skip tests in mixed version tests Skip tests in mixed version tests because feature flag rabbitmq_4.0.0 is needed for the new #modify{} Ra command being sent to quorum queues. --------- Co-authored-by: David Ansari <david.ansari@gmx.de> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2024-08-08 15:48:27 +08:00
- rabbit_fifo_v3
- rabbit_file
- rabbit_global_counters
- rabbit_guid
- rabbit_health_check
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
- rabbit_khepri
- rabbit_limiter
- rabbit_log_channel
- rabbit_log_connection
- rabbit_log_mirroring
- rabbit_log_prelaunch
- rabbit_log_queue
- rabbit_log_tail
- rabbit_logger_exchange_h
- rabbit_maintenance
Move plugin rabbitmq-message-timestamp to the core As reported in https://groups.google.com/g/rabbitmq-users/c/x8ACs4dBlkI/ plugins that implement rabbit_channel_interceptor break with Native MQTT in 3.12 because Native MQTT does not use rabbit_channel anymore. Specifically, these plugins don't work anymore in 3.12 when sending a message from an MQTT publisher to an AMQP 0.9.1 consumer. Two of these plugins are https://github.com/rabbitmq/rabbitmq-message-timestamp and https://github.com/rabbitmq/rabbitmq-routing-node-stamp This commit moves both plugins into rabbitmq-server. Therefore, these plugins are deprecated starting in 3.12. Instead of using these plugins, the user gets the same behaviour by configuring rabbitmq.conf as follows: ``` incoming_message_interceptors.set_header_timestamp.overwrite = false incoming_message_interceptors.set_header_routing_node.overwrite = false ``` While both plugins were incompatible to be used together, this commit allows setting both headers. We name the top level configuration key `incoming_message_interceptors` because only incoming messages are intercepted. Currently, only `set_header_timestamp` and `set_header_routing_node` are supported. (We might support more in the future.) Both can set `overwrite` to `false` or `true`. The meaning of `overwrite` is the same as documented in https://github.com/rabbitmq/rabbitmq-message-timestamp#always-overwrite-timestamps i.e. whether headers should be overwritten if they are already present in the message. Both `set_header_timestamp` and `set_header_routing_node` behave exactly to plugins `rabbitmq-message-timestamp` and `rabbitmq-routing-node-stamp`, respectively. Upon node boot, the configuration is put into persistent_term to not cause any performance penalty in the default case where these settings are disabled. The channel and MQTT connection process will intercept incoming messages and - if configured - add the desired AMQP 0.9.1 headers. For now, this allows using Native MQTT in 3.12 with the old plugins behaviour. In the future, once "message containers" are implemented, we can think about more generic message interceptors where plugins can be written to modify arbitrary headers or message contents for various protocols. Likewise, in the future, once MQTT 5.0 is implemented, we can think about an MQTT connection interceptor which could function similar to a `rabbit_channel_interceptor` allowing to modify any MQTT packet.
2023-05-12 22:12:50 +08:00
- rabbit_message_interceptor
- rabbit_metrics
- rabbit_mirror_queue_misc
- rabbit_mnesia
- rabbit_msg_store
- rabbit_msg_store_gc
- rabbit_networking
- rabbit_networking_store
- rabbit_node_monitor
- rabbit_nodes
- rabbit_observer_cli
- rabbit_observer_cli_classic_queues
2023-05-19 00:25:08 +08:00
- rabbit_observer_cli_quorum_queues
- rabbit_osiris_metrics
- rabbit_parameter_validation
- rabbit_peer_discovery
- rabbit_peer_discovery_classic_config
- rabbit_peer_discovery_dns
- rabbit_plugins
- rabbit_policies
- rabbit_policy
- rabbit_policy_merge_strategy
- rabbit_prelaunch_cluster
- rabbit_prelaunch_enabled_plugins_file
- rabbit_prelaunch_feature_flags
- rabbit_prelaunch_logging
- rabbit_priority_queue
- rabbit_process
- rabbit_queue_consumers
- rabbit_queue_decorator
- rabbit_queue_index
- rabbit_queue_location
- rabbit_queue_type
- rabbit_queue_type_util
- rabbit_quorum_memory_manager
- rabbit_quorum_queue
- rabbit_quorum_queue_periodic_membership_reconciliation
- rabbit_ra_registry
- rabbit_ra_systems
- rabbit_reader
- rabbit_recovery_terms
- rabbit_release_series
- rabbit_restartable_sup
- rabbit_router
- rabbit_runtime_parameters
- rabbit_ssl
- rabbit_stream_coordinator
- rabbit_stream_queue
- rabbit_stream_sac_coordinator
- rabbit_sup
- rabbit_sysmon_handler
- rabbit_sysmon_minder
- rabbit_table
- rabbit_time_travel_dbg
- rabbit_trace
- rabbit_tracking
- rabbit_tracking_store
- rabbit_upgrade_preparation
- rabbit_uri
- rabbit_variable_queue
- rabbit_version
- rabbit_vhost
- rabbit_vhost_limit
- rabbit_vhost_msg_store
- rabbit_vhost_process
- rabbit_vhost_sup
- rabbit_vhost_sup_sup
- rabbit_vhost_sup_wrapper
- rabbit_vhosts
- rabbit_vm
- supervised_lifecycle
- tcp_listener
- tcp_listener_sup
- term_to_binary_compat
- vhost
rabbit_common:
- app_utils
- code_version
- credit_flow
- delegate
- delegate_sup
- file_handle_cache
- gen_server2
- mirrored_supervisor_locks
- mnesia_sync
- pmon
- priority_queue
- rabbit_amqp_connection
- rabbit_amqqueue_common
- rabbit_auth_backend_dummy
- rabbit_auth_mechanism
- rabbit_authn_backend
- rabbit_authz_backend
- rabbit_basic_common
- rabbit_binary_generator
- rabbit_binary_parser
- rabbit_cert_info
- rabbit_channel_common
- rabbit_command_assembler
- rabbit_control_misc
- rabbit_core_metrics
- rabbit_data_coercion
- rabbit_date_time
- rabbit_env
- rabbit_error_logger_handler
- rabbit_event
- rabbit_framing
- rabbit_framing_amqp_0_8
- rabbit_framing_amqp_0_9_1
- rabbit_heartbeat
- rabbit_http_util
- rabbit_json
- rabbit_log
- rabbit_misc
- rabbit_net
- rabbit_nodes_common
- rabbit_numerical
- rabbit_password
- rabbit_password_hashing
- rabbit_password_hashing_md5
- rabbit_password_hashing_sha256
- rabbit_password_hashing_sha512
- rabbit_pbe
- rabbit_peer_discovery_backend
- rabbit_policy_validator
- rabbit_queue_collector
- rabbit_registry
- rabbit_registry_class
- rabbit_resource_monitor_misc
Support AMQP 1.0 natively ## What Similar to Native MQTT in #5895, this commits implements Native AMQP 1.0. By "native", we mean do not proxy via AMQP 0.9.1 anymore. ## Why Native AMQP 1.0 comes with the following major benefits: 1. Similar to Native MQTT, this commit provides better throughput, latency, scalability, and resource usage for AMQP 1.0. See https://blog.rabbitmq.com/posts/2023/03/native-mqtt for native MQTT improvements. See further below for some benchmarks. 2. Since AMQP 1.0 is not limited anymore by the AMQP 0.9.1 protocol, this commit allows implementing more AMQP 1.0 features in the future. Some features are already implemented in this commit (see next section). 3. Simpler, better understandable, and more maintainable code. Native AMQP 1.0 as implemented in this commit has the following major benefits compared to AMQP 0.9.1: 4. Memory and disk alarms will only stop accepting incoming TRANSFER frames. New connections can still be created to consume from RabbitMQ to empty queues. 5. Due to 4. no need anymore for separate connections for publishers and consumers as we currently recommended for AMQP 0.9.1. which potentially halves the number of physical TCP connections. 6. When a single connection sends to multiple target queues, a single slow target queue won't block the entire connection. Publisher can still send data quickly to all other target queues. 7. A publisher can request whether it wants publisher confirmation on a per-message basis. In AMQP 0.9.1 publisher confirms are configured per channel only. 8. Consumers can change their "prefetch count" dynamically which isn't possible in our AMQP 0.9.1 implementation. See #10174 9. AMQP 1.0 is an extensible protocol This commit also fixes dozens of bugs present in the AMQP 1.0 plugin in RabbitMQ 3.x - most of which cannot be backported due to the complexity and limitations of the old 3.x implementation. This commit contains breaking changes and is therefore targeted for RabbitMQ 4.0. ## Implementation details 1. Breaking change: With Native AMQP, the behaviour of ``` Convert AMQP 0.9.1 message headers to application properties for an AMQP 1.0 consumer amqp1_0.convert_amqp091_headers_to_app_props = false | true (default false) Convert AMQP 1.0 Application Properties to AMQP 0.9.1 headers amqp1_0.convert_app_props_to_amqp091_headers = false | true (default false) ``` will break because we always convert according to the message container conversions. For example, AMQP 0.9.1 x-headers will go into message-annotations instead of application properties. Also, `false` won’t be respected since we always convert the headers with message containers. 2. Remove rabbit_queue_collector rabbit_queue_collector is responsible for synchronously deleting exclusive queues. Since the AMQP 1.0 plugin never creates exclusive queues, rabbit_queue_collector doesn't need to be started in the first place. This will save 1 Erlang process per AMQP 1.0 connection. 3. 7 processes per connection + 1 process per session in this commit instead of 7 processes per connection + 15 processes per session in 3.x Supervision hierarchy got re-designed. 4. Use 1 writer process per AMQP 1.0 connection AMQP 0.9.1 uses a separate rabbit_writer Erlang process per AMQP 0.9.1 channel. Prior to this commit, AMQP 1.0 used a separate rabbit_amqp1_0_writer process per AMQP 1.0 session. Advantage of single writer proc per session (prior to this commit): * High parallelism for serialising packets if multiple sessions within a connection write heavily at the same time. This commit uses a single writer process per AMQP 1.0 connection that is shared across all AMQP 1.0 sessions. Advantages of single writer proc per connection (this commit): * Lower memory usage with hundreds of thousands of AMQP 1.0 sessions * Less TCP and IP header overhead given that the single writer process can accumulate across all sessions bytes before flushing the socket. In other words, this commit decides that a reader / writer process pair per AMQP 1.0 connection is good enough for bi-directional TRANSFER flows. Having a writer per session is too heavy. We still ensure high throughput by having separate reader, writer, and session processes. 5. Transform rabbit_amqp1_0_writer into gen_server Why: Prior to this commit, when clicking on the AMQP 1.0 writer process in observer, the process crashed. Instead of handling all these debug messages of the sys module, it's better to implement a gen_server. There is no advantage of using a special OTP process over gen_server for the AMQP 1.0 writer. gen_server also provides cleaner format status output. How: Message callbacks return a timeout of 0. After all messages in the inbox are processed, the timeout message is handled by flushing any pending bytes. 6. Remove stats timer from writer AMQP 1.0 connections haven't emitted any stats previously. 7. When there are contiguous queue confirmations in the session process mailbox, batch them. When the confirmations are sent to the publisher, a single DISPOSITION frame is sent for contiguously confirmed delivery IDs. This approach should be good enough. However it's sub optimal in scenarios where contiguous delivery IDs that need confirmations are rare, for example: * There are multiple links in the session with different sender settlement modes and sender publishes across these links interleaved. * sender settlement mode is mixed and sender publishes interleaved settled and unsettled TRANSFERs. 8. Introduce credit API v2 Why: The AMQP 0.9.1 credit extension which is to be removed in 4.0 was poorly designed since basic.credit is a synchronous call into the queue process blocking the entire AMQP 1.0 session process. How: Change the interactions between queue clients and queue server implementations: * Clients only request a credit reply if the FLOW's `echo` field is set * Include all link flow control state held by the queue process into a new credit_reply queue event: * `available` after the queue sends any deliveries * `link-credit` after the queue sends any deliveries * `drain` which allows us to combine the old queue events send_credit_reply and send_drained into a single new queue event credit_reply. * Include the consumer tag into the credit_reply queue event such that the AMQP 1.0 session process can process any credit replies asynchronously. Link flow control state `delivery-count` also moves to the queue processes. The new interactions are hidden behind feature flag credit_api_v2 to allow for rolling upgrades from 3.13 to 4.0. 9. Use serial number arithmetic in quorum queues and session process. 10. Completely bypass the rabbit_limiter module for AMQP 1.0 flow control. The goal is to eventually remove the rabbit_limiter module in 4.0 since AMQP 0.9.1 global QoS will be unsupported in 4.0. This commit lifts the AMQP 1.0 link flow control logic out of rabbit_limiter into rabbit_queue_consumers. 11. Fix credit bug for streams: AMQP 1.0 settlements shouldn't top up link credit, only FLOW frames should top up link credit. 12. Allow sender settle mode unsettled for streams since AMQP 1.0 acknowledgements to streams are no-ops (currently). 13. Fix AMQP 1.0 client bugs Auto renewing credits should not be related to settling TRANSFERs. Remove field link_credit_unsettled as it was wrong and confusing. Prior to this commit auto renewal did not work when the sender uses sender settlement mode settled. 14. Fix AMQP 1.0 client bugs The wrong outdated Link was passed to function auto_flow/2 15. Use osiris chunk iterator Only hold messages of uncompressed sub batches in memory if consumer doesn't have sufficient credits. Compressed sub batches are skipped for non Stream protocol consumers. 16. Fix incoming link flow control Always use confirms between AMQP 1.0 queue clients and queue servers. As already done internally by rabbit_fifo_client and rabbit_stream_queue, use confirms for classic queues as well. 17. Include link handle into correlation when publishing messages to target queues such that session process can correlate confirms from target queues to incoming links. 18. Only grant more credits to publishers if publisher hasn't sufficient credits anymore and there are not too many unconfirmed messages on the link. 19. Completely ignore `block` and `unblock` queue actions and RabbitMQ credit flow between classic queue process and session process. 20. Link flow control is independent between links. A client can refer to a queue or to an exchange with multiple dynamically added target queues. Multiple incoming links can also fan in to the same queue. However the link topology looks like, this commit ensures that each link is only granted more credits if that link isn't overloaded. 21. A connection or a session can send to many different queues. In AMQP 0.9.1, a single slow queue will lead to the entire channel, and then entire connection being blocked. This commit makes sure that a single slow queue from one link won't slow down sending on other links. For example, having link A sending to a local classic queue and link B sending to 5 replica quorum queue, link B will naturally grant credits slower than link A. So, despite the quorum queue being slower in confirming messages, the same AMQP 1.0 connection and session can still pump data very fast into the classic queue. 22. If cluster wide memory or disk alarm occurs. Each session sends a FLOW with incoming-window to 0 to sending client. If sending clients don’t obey, force disconnect the client. If cluster wide memory alarm clears: Each session resumes with a FLOW defaulting to initial incoming-window. 23. All operations apart of publishing TRANSFERS to RabbitMQ can continue during cluster wide alarms, specifically, attaching consumers and consuming, i.e. emptying queues. There is no need for separate AMQP 1.0 connections for publishers and consumers as recommended in our AMQP 0.9.1 implementation. 24. Flow control summary: * If queue becomes bottleneck, that’s solved by slowing down individual sending links (AMQP 1.0 link flow control). * If session becomes bottleneck (more unlikely), that’s solved by AMQP 1.0 session flow control. * If connection becomes bottleneck, it naturally won’t read fast enough from the socket causing TCP backpressure being applied. Nowhere will RabbitMQ internal credit based flow control (i.e. module credit_flow) be used on the incoming AMQP 1.0 message path. 25. Register AMQP sessions Prefer local-only pg over our custom pg_local implementation as pg is a better process group implementation than pg_local. pg_local was identified as bottleneck in tests where many MQTT clients were disconnected at once. 26. Start a local-only pg when Rabbit boots: > A scope can be kept local-only by using a scope name that is unique cluster-wide, e.g. the node name: > pg:start_link(node()). Register AMQP 1.0 connections and sessions with pg. In future we should remove pg_local and instead use the new local-only pg for all registered processes such as AMQP 0.9.1 connections and channels. 27. Requeue messages if link detached Although the spec allows to settle delivery IDs on detached links, RabbitMQ does not respect the 'closed' field of the DETACH frame and therefore handles every DETACH frame as closed. Since the link is closed, we expect every outstanding delivery to be requeued. In addition to consumer cancellation, detaching a link therefore causes in flight deliveries to be requeued. Note that this behaviour is different from merely consumer cancellation in AMQP 0.9.1: "After a consumer is cancelled there will be no future deliveries dispatched to it. Note that there can still be "in flight" deliveries dispatched previously. Cancelling a consumer will neither discard nor requeue them." [https://www.rabbitmq.com/consumers.html#unsubscribing] An AMQP receiver can first drain, and then detach to prevent "in flight" deliveries 28. Init AMQP session with BEGIN frame Similar to how there can't be an MQTT processor without a CONNECT frame, there can't be an AMQP session without a BEGIN frame. This allows having strict dialyzer types for session flow control fields (i.e. not allowing 'undefined'). 29. Move serial_number to AMQP 1.0 common lib such that it can be used by both AMQP 1.0 server and client 30. Fix AMQP client to do serial number arithmetic. 31. AMQP client: Differentiate between delivery-id and transfer-id for better understandability. 32. Fix link flow control in classic queues This commit fixes ``` java -jar target/perf-test.jar -ad false -f persistent -u cq -c 3000 -C 1000000 -y 0 ``` followed by ``` ./omq -x 0 amqp -T /queue/cq -D 1000000 --amqp-consumer-credits 2 ``` Prior to this commit, (and on RabbitMQ 3.x) the consuming would halt after around 8 - 10,000 messages. The bug was that in flight messages from classic queue process to session process were not taken into account when topping up credit to the classic queue process. Fixes #2597 The solution to this bug (and a much cleaner design anyway independent of this bug) is that queues should hold all link flow control state including the delivery-count. Hence, when credit API v2 is used the delivery-count will be held by the classic queue process, quorum queue process, and stream queue client instead of managing the delivery-count in the session. 33. The double level crediting between (a) session process and rabbit_fifo_client, and (b) rabbit_fifo_client and rabbit_fifo was removed. Therefore, instead of managing 3 separate delivery-counts (i. session, ii. rabbit_fifo_client, iii. rabbit_fifo), only 1 delivery-count is used in rabbit_fifo. This is a big simplification. 34. This commit fixes quorum queues without bumping the machine version nor introducing new rabbit_fifo commands. Whether credit API v2 is used is solely determined at link attachment time depending on whether feature flag credit_api_v2 is enabled. Even when that feature flag will be enabled later on, this link will keep using credit API v1 until detached (or the node is shut down). Eventually, after feature flag credit_api_v2 has been enabled and a subsequent rolling upgrade, all links will use credit API v2. This approach is safe and simple. The 2 alternatives to move delivery-count from the session process to the queue processes would have been: i. Explicit feature flag credit_api_v2 migration function * Can use a gen_server:call and only finish migration once all delivery-counts were migrated. Cons: * Extra new message format just for migration is required. * Risky as migration will fail if a target queue doesn’t reply. ii. Session always includes DeliveryCountSnd when crediting to the queue: Cons: * 2 delivery counts will be hold simultaneously in session proc and queue proc; could be solved by deleting the session proc’s delivery-count for credit-reply * What happens if the receiver doesn’t provide credit for a very long time? Is that a problem? 35. Support stream filtering in AMQP 1.0 (by @acogoluegnes) Use the x-stream-filter-value message annotation to carry the filter value in a published message. Use the rabbitmq:stream-filter and rabbitmq:stream-match-unfiltered filters when creating a receiver that wants to filter out messages from a stream. 36. Remove credit extension from AMQP 0.9.1 client 37. Support maintenance mode closing AMQP 1.0 connections. 38. Remove AMQP 0.9.1 client dependency from AMQP 1.0 implementation. 39. Move AMQP 1.0 plugin to the core. AMQP 1.0 is enabled by default. The old rabbitmq_amqp1_0 plugin will be kept as a no-op plugin to prevent deployment tools from failing that execute: ``` rabbitmq-plugins enable rabbitmq_amqp1_0 rabbitmq-plugins disable rabbitmq_amqp1_0 ``` 40. Breaking change: Remove CLI command `rabbitmqctl list_amqp10_connections`. Instead, list both AMQP 0.9.1 and AMQP 1.0 connections in `list_connections`: ``` rabbitmqctl list_connections protocol Listing connections ... protocol {1, 0} {0,9,1} ``` ## Benchmarks ### Throughput & Latency Setup: * Single node Ubuntu 22.04 * Erlang 26.1.1 Start RabbitMQ: ``` make run-broker PLUGINS="rabbitmq_management rabbitmq_amqp1_0" FULL=1 RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+S 3" ``` Predeclare durable classic queue cq1, durable quorum queue qq1, durable stream queue sq1. Start client: https://github.com/ssorj/quiver https://hub.docker.com/r/ssorj/quiver/tags (digest 453a2aceda64) ``` docker run -it --rm --add-host host.docker.internal:host-gateway ssorj/quiver:latest bash-5.1# quiver --version quiver 0.4.0-SNAPSHOT ``` 1. Classic queue ``` quiver //host.docker.internal//amq/queue/cq1 --durable --count 1m --duration 10m --body-size 12 --credit 1000 ``` This commit: ``` Count ............................................. 1,000,000 messages Duration ............................................... 73.8 seconds Sender rate .......................................... 13,548 messages/s Receiver rate ........................................ 13,547 messages/s End-to-end rate ...................................... 13,547 messages/s Latencies by percentile: 0% ........ 0 ms 90.00% ........ 9 ms 25% ........ 2 ms 99.00% ....... 14 ms 50% ........ 4 ms 99.90% ....... 17 ms 100% ....... 26 ms 99.99% ....... 24 ms ``` RabbitMQ 3.x (main branch as of 30 January 2024): ``` ---------------------- Sender ----------------------- --------------------- Receiver ---------------------- -------- Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Lat [ms] ----------------------------------------------------- ----------------------------------------------------- -------- 2.1 130,814 65,342 6 73.6 2.1 3,217 1,607 0 8.0 511 4.1 163,580 16,367 2 74.1 4.1 3,217 0 0 8.0 0 6.1 229,114 32,767 3 74.1 6.1 3,217 0 0 8.0 0 8.1 261,880 16,367 2 74.1 8.1 67,874 32,296 8 8.2 7,662 10.1 294,646 16,367 2 74.1 10.1 67,874 0 0 8.2 0 12.1 360,180 32,734 3 74.1 12.1 67,874 0 0 8.2 0 14.1 392,946 16,367 3 74.1 14.1 68,604 365 0 8.2 12,147 16.1 458,480 32,734 3 74.1 16.1 68,604 0 0 8.2 0 18.1 491,246 16,367 2 74.1 18.1 68,604 0 0 8.2 0 20.1 556,780 32,767 4 74.1 20.1 68,604 0 0 8.2 0 22.1 589,546 16,375 2 74.1 22.1 68,604 0 0 8.2 0 receiver timed out 24.1 622,312 16,367 2 74.1 24.1 68,604 0 0 8.2 0 quiver: error: PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/cq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-otujr23y' returned non-zero exit status 1. Traceback (most recent call last): File "/usr/local/lib/quiver/python/quiver/pair.py", line 144, in run _plano.wait(receiver, check=True) File "/usr/local/lib/quiver/python/plano/main.py", line 1243, in wait raise PlanoProcessError(proc) plano.main.PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/cq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-otujr23y' returned non-zero exit status 1. ``` 2. Quorum queue: ``` quiver //host.docker.internal//amq/queue/qq1 --durable --count 1m --duration 10m --body-size 12 --credit 1000 ``` This commit: ``` Count ............................................. 1,000,000 messages Duration .............................................. 101.4 seconds Sender rate ........................................... 9,867 messages/s Receiver rate ......................................... 9,868 messages/s End-to-end rate ....................................... 9,865 messages/s Latencies by percentile: 0% ....... 11 ms 90.00% ....... 23 ms 25% ....... 15 ms 99.00% ....... 28 ms 50% ....... 18 ms 99.90% ....... 33 ms 100% ....... 49 ms 99.99% ....... 47 ms ``` RabbitMQ 3.x: ``` ---------------------- Sender ----------------------- --------------------- Receiver ---------------------- -------- Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Lat [ms] ----------------------------------------------------- ----------------------------------------------------- -------- 2.1 130,814 65,342 9 69.9 2.1 18,430 9,206 5 7.6 1,221 4.1 163,580 16,375 5 70.2 4.1 18,867 218 0 7.6 2,168 6.1 229,114 32,767 6 70.2 6.1 18,867 0 0 7.6 0 8.1 294,648 32,734 7 70.2 8.1 18,867 0 0 7.6 0 10.1 360,182 32,734 6 70.2 10.1 18,867 0 0 7.6 0 12.1 425,716 32,767 6 70.2 12.1 18,867 0 0 7.6 0 receiver timed out 14.1 458,482 16,367 5 70.2 14.1 18,867 0 0 7.6 0 quiver: error: PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/qq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-b1gcup43' returned non-zero exit status 1. Traceback (most recent call last): File "/usr/local/lib/quiver/python/quiver/pair.py", line 144, in run _plano.wait(receiver, check=True) File "/usr/local/lib/quiver/python/plano/main.py", line 1243, in wait raise PlanoProcessError(proc) plano.main.PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/qq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-b1gcup43' returned non-zero exit status 1. ``` 3. Stream: ``` quiver-arrow send //host.docker.internal//amq/queue/sq1 --durable --count 1m -d 10m --summary --verbose ``` This commit: ``` Count ............................................. 1,000,000 messages Duration ................................................ 8.7 seconds Message rate ........................................ 115,154 messages/s ``` RabbitMQ 3.x: ``` Count ............................................. 1,000,000 messages Duration ............................................... 21.2 seconds Message rate ......................................... 47,232 messages/s ``` ### Memory usage Start RabbitMQ: ``` ERL_MAX_PORTS=3000000 RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+P 3000000 +S 6" make run-broker PLUGINS="rabbitmq_amqp1_0" FULL=1 RABBITMQ_CONFIG_FILE="rabbitmq.conf" ``` ``` /bin/cat rabbitmq.conf tcp_listen_options.sndbuf = 2048 tcp_listen_options.recbuf = 2048 vm_memory_high_watermark.relative = 0.95 vm_memory_high_watermark_paging_ratio = 0.95 loopback_users = none ``` Create 50k connections with 2 sessions per connection, i.e. 100k session in total: ```go package main import ( "context" "log" "time" "github.com/Azure/go-amqp" ) func main() { for i := 0; i < 50000; i++ { conn, err := amqp.Dial(context.TODO(), "amqp://nuc", &amqp.ConnOptions{SASLType: amqp.SASLTypeAnonymous()}) if err != nil { log.Fatal("dialing AMQP server:", err) } _, err = conn.NewSession(context.TODO(), nil) if err != nil { log.Fatal("creating AMQP session:", err) } _, err = conn.NewSession(context.TODO(), nil) if err != nil { log.Fatal("creating AMQP session:", err) } } log.Println("opened all connections") time.Sleep(5 * time.Hour) } ``` This commit: ``` erlang:memory(). [{total,4586376480}, {processes,4025898504}, {processes_used,4025871040}, {system,560477976}, {atom,1048841}, {atom_used,1042841}, {binary,233228608}, {code,21449982}, {ets,108560464}] erlang:system_info(process_count). 450289 ``` 7 procs per connection + 1 proc per session. (7 + 2*1) * 50,000 = 450,000 procs RabbitMQ 3.x: ``` erlang:memory(). [{total,15168232704}, {processes,14044779256}, {processes_used,14044755120}, {system,1123453448}, {atom,1057033}, {atom_used,1052587}, {binary,236381264}, {code,21790238}, {ets,391423744}] erlang:system_info(process_count). 1850309 ``` 7 procs per connection + 15 per session (7 + 2*15) * 50,000 = 1,850,000 procs 50k connections + 100k session require with this commit: 4.5 GB in RabbitMQ 3.x: 15 GB ## Future work 1. More efficient parser and serializer 2. TODO in mc_amqp: Do not store the parsed message on disk. 3. Implement both AMQP HTTP extension and AMQP management extension to allow AMQP clients to create RabbitMQ objects (queues, exchanges, ...).
2023-07-21 18:29:07 +08:00
- rabbit_routing_parser
- rabbit_runtime
- rabbit_runtime_parameter
- rabbit_semver
- rabbit_semver_parser
- rabbit_ssl_options
- rabbit_types
- rabbit_writer
- supervisor2
- vm_memory_monitor
- worker_pool
- worker_pool_sup
- worker_pool_worker
Enable AMQP 1.0 clients to manage topologies ## What? * Allow AMQP 1.0 clients to dynamically create and delete RabbitMQ topologies (exchanges, queues, bindings). * Provide an Erlang AMQP 1.0 client that manages topologies. ## Why? Today, RabbitMQ topologies can be created via: * [Management HTTP API](https://www.rabbitmq.com/docs/management#http-api) (including Management UI and [messaging-topology-operator](https://github.com/rabbitmq/messaging-topology-operator)) * [Definition Import](https://www.rabbitmq.com/docs/definitions#import) * AMQP 0.9.1 clients Up to RabbitMQ 3.13 the RabbitMQ AMQP 1.0 plugin auto creates queues and bindings depending on the terminus [address format](https://github.com/rabbitmq/rabbitmq-server/tree/v3.13.x/deps/rabbitmq_amqp1_0#routing-and-addressing). Such implicit creation of topologies is limiting and obscure. For some address formats, queues will be created, but not deleted. Some of RabbitMQ's success is due to its flexible routing topologies that AMQP 0.9.1 clients can create and delete dynamically. This commit allows dynamic management of topologies for AMQP 1.0 clients. This commit builds on top of Native AMQP 1.0 (PR #9022) and will be available in RabbitMQ 4.0. ## How? This commits adds the following management operations for AMQP 1.0 clients: * declare queue * delete queue * purge queue * bind queue to exchange * unbind queue from exchange * declare exchange * delete exchange * bind exchange to exchange * unbind exchange from exchange Hence, at least the AMQP 0.9.1 management operations are supported for AMQP 1.0 clients. In addition the operation * get queue is provided which - similar to `declare queue` - returns queue information including the current leader and replicas. This allows clients to publish or consume locally on the node that hosts the queue. Compared to AMQP 0.9.1 whose commands and command fields are fixed, the new AMQP Management API is extensible: New operations and new fields can easily be added in the future. There are different design options how management operations could be supported for AMQP 1.0 clients: 1. Use a special exchange type as done in https://github.com/rabbitmq/rabbitmq-management-exchange This has the advantage that any protocol client (e.g. also STOMP clients) could dynamically manage topologies. However, a special exchange type is the wrong abstraction. 2. Clients could send "special" messages with special headers that the broker interprets. This commit decided for a variation of the 2nd option using a more standardized way by re-using a subest of the following latest AMQP 1.0 extension specifications: * [AMQP Request-Response Messaging with Link Pairing Version 1.0 - Committee Specification 01](https://docs.oasis-open.org/amqp/linkpair/v1.0/cs01/linkpair-v1.0-cs01.html) (February 2021) * [HTTP Semantics and Content over AMQP Version 1.0 - Working Draft 06](https://groups.oasis-open.org/higherlogic/ws/public/document?document_id=65571) (July 2019) * [AMQP Management Version 1.0 - Working Draft 16](https://groups.oasis-open.org/higherlogic/ws/public/document?document_id=65575) (July 2019) An important goal is to keep the interaction between AMQP 1.0 client and RabbitMQ simple to increase usage, development and adoptability of future RabbitMQ AMQP 1.0 client library wrappers. The AMQP 1.0 client has to create a link pair to the special `/management` node. This allows the client to send and receive from the management node. Similar to AMQP 0.9.1, there is no need for a reply queue since the reply will be sent directly to the client. Requests and responses are modelled via HTTP, but sent via AMQP using the `HTTP Semantics and Content over AMQP` extension (henceforth `HTTP over AMQP` extension). This commit tries to follow the `HTTP over AMQP` extension as much as possible but deviates where this draft spec doesn't make sense. The projected mode §4.1 is used as opposed to tunneled mode §4.2. A named relay `/management` is used (§6.3) where the message field `to` is the URL. Deviations are * §3.1 mandates that URIs are not encoded in an AMQP message. However, we percent encode URIs in the AMQP message. Otherwise there is for example no way to distinguish a `/` in a queue name from the URI path separator `/`. * §4.1.4 mandates a data section. This commit uses an amqp-value section as it's a better fit given that the content is AMQP encoded data. Using an HTTP API allows for a common well understood interface and future extensibility. Instead of re-using the current RabbitMQ HTTP API, this commit uses a new HTTP API (let's call it v2) which could be used as a future API for plain HTTP clients. ### HTTP API v1 The current HTTP API (let's call it v1) is **not** used since v1 comes with a couple of weaknesses: 1. Deep level of nesting becomes confusing and difficult to manage. Examples of deep nesting in v1: ``` /api/bindings/vhost/e/source/e/destination/props /api/bindings/vhost/e/exchange/q/queue/props ``` 2. Redundant endpoints returning the same resources v1 has 9 endpoints to list binding(s): ``` /api/exchanges/vhost/name/bindings/source /api/exchanges/vhost/name/bindings/destination /api/queues/vhost/name/bindings /api/bindings /api/bindings/vhost /api/bindings/vhost/e/exchange/q/queue /api/bindings/vhost/e/exchange/q/queue/props /api/bindings/vhost/e/source/e/destination /api/bindings/vhost/e/source/e/destination/props ``` 3. Verbs in path names Path names should be nouns instead. v1 contains verbs: ``` /api/queues/vhost/name/get /api/exchanges/vhost/name/publish ``` ### AMQP Management extension Only few aspects of the AMQP Management extension are used. The central idea of the AMQP management spec is **dynamic discovery** such that broker independent AMQP 1.0 clients can discover objects, types, operations, and HTTP endpoints of specific brokers. In fact, clients are only conformant if: > All request addresses are dynamically discovered starting from the discovery document. > A requesting container MUST NOT use fixed assumptions about the addressing structure of the management API. While this is a nice and powerful idea, no AMQP 1.0 client and no AMQP 1.0 server implement the latest AMQP 1.0 management spec from 2019, partly presumably due to its complexity. Therefore, the idea of such dynamic discovery has failed to be implemented in practice. The AMQP management spec mandates that the management endpoint returns a discovery document containing broker specific collections, types, configuration, and operations including their endpoints. The API endpoints of the AMQP management spec are therefore all designed around dynamic discovery. For example, to create either a queue or an exchange, the client has to ``` POST /$management/entities ``` which shows that the entities collection acts as a generic factory, see section 2.2. The server will then create the resource and reply with a location header containing a URI pointing to the resource. For RabbitMQ, we don’t need such a generic factory to create queues or exchanges. To list bindings for a queue Q1, the spec suggests ``` GET /$management/Queues/Q1/$management/entities ``` which again shows the generic entities endpoint as well as a `$management` endpoint under Q1 to allow a queue to return a discovery document. For RabbitMQ, we don’t need such generic endpoints and discovery documents. Given we aim for our own thin RabbitMQ AMQP 1.0 client wrapper libraries which expose the RabbitMQ model to the developer, we can directly use fixed HTTP endpoint assumptions in our RabbitMQ specific libraries. This is by far simpler than using the dynamic endpoints of the management spec. Simplicity leads to higher adoption and enables more developers to write RabbitMQ AMQP 1.0 client library wrappers. The AMQP Management extension also suffers from deep level of nesting in paths Examples: ``` /$management/Queues/Q1/$management/entities /$management/Queues/Q1/Bindings/Binding1 ``` as well as verbs in path names: Section 7.1.4 suggests using verbs in path names, for example “purge”, due to the dynamic operations discovery document. ### HTTP API v2 This commit introduces a new HTTP API v2 following best practices. It could serve as a future API for plain HTTP clients. This commit and RabbitMQ 4.0 will only implement a minimal set of HTTP API v2 endpoints and only for HTTP over AMQP. In other words, the existing HTTP API v1 Cowboy handlers will continue to be used for all plain HTTP requests in RabbitMQ 4.0 and will remain untouched for RabbitMQ 4.0. Over time, after 4.0 shipped, we could ship a pure HTTP API implementation for HTTP API v2. Hence, the new HTTP API v2 endpoints for HTTP over AMQP should be designed such that they can be re-used in the future for a pure HTTP implementation. The minimal set of endpoints for RabbitMQ 4.0 are: `` GET / PUT / DELETE /vhosts/:vhost/queues/:queue ``` read, create, delete a queue ``` DELETE /vhosts/:vhost/queues/:queue/messages ``` purges a queue ``` GET / DELETE /vhosts/:vhost/bindings/:binding ``` read, delete bindings where `:binding` is a binding ID of the following path segment: ``` src=e1;dstq=q2;key=my-key;args= ``` Binding arguments `args` has an empty value by default, i.e. there are no binding arguments. If the binding includes binding arguments, `args` will be an Erlang portable term hash provided by the server similar to what’s provided in HTTP API v1 today. Alternatively, we could use an arguments scheme of: ``` args=k1,utf8,v1&k2,uint,3 ``` However, such a scheme leads to long URIs when there are many binding arguments. Note that it’s perfectly fine for URI producing applications to include URI reserved characters `=` / `;` / `,` / `$` in a path segment. To create a binding, the client therefore needs to POST to a bindings factory URI: ``` POST /vhosts/:vhost/bindings ``` To list all bindings between a source exchange e1 and destination exchange e2 with binding key k1: ``` GET /vhosts/:vhost/bindings?src=e1&dste=e2&key=k1 ``` This endpoint will be called by the RabbitMQ AMQP 1.0 client library to unbind a binding with non-empty binding arguments to get the binding ID before invoking a ``` DELETE /vhosts/:vhost/bindings/:binding ``` In future, after RabbitMQ 4.0 shipped, new API endpoints could be added. The following is up for discussion and is only meant to show the clean and simple design of HTTP API v2. Bindings endpoint can be queried as follows: to list all bindings for a given source exchange e1: ``` GET /vhosts/:vhost/bindings?src=e1 ``` to list all bindings for a given destination queue q1: ``` GET /vhosts/:vhost/bindings?dstq=q1 ``` to list all bindings between a source exchange e1 and destination queue q1: ``` GET /vhosts/:vhost/bindings?src=e1&dstq=q1 ``` multiple bindings between source exchange e1 and destination queue q1 could be deleted at once as follows: ``` DELETE /vhosts/:vhost/bindings?src=e1&dstq=q1 ``` GET could be supported globally across all vhosts: ``` /exchanges /queues /bindings ``` Publish a message: ``` POST /vhosts/:vhost/queues/:queue/messages ``` Consume or peek a message (depending on query parameters): ``` GET /vhosts/:vhost/queues/:queue/messages ``` Note that the AMQP 1.0 client omits the `/vhost/:vhost` path prefix. Since an AMQP connection belongs to a single vhost, there is no need to additionally include the vhost in every HTTP request. Pros of HTTP API v2: 1. Low level of nesting Queues, exchanges, bindings are top level entities directly under vhosts. Although the HTTP API doesn’t have to reflect how resources are stored in the database, v2 does nicely reflect the Khepri tree structure. 2. Nouns instead of verbs HTTP API v2 is very simple to read and understand as shown by ``` POST /vhosts/:vhost/queues/:queue/messages to post messages, i.e. publish to a queue. GET /vhosts/:vhost/queues/:queue/messages to get messages, i.e. consume or peek from a queue. DELETE /vhosts/:vhost/queues/:queue/messages to delete messages, i.e. purge a queue. ``` A separate new HTTP API v2 allows us to ship only handlers for HTTP over AMQP for RabbitMQ 4.0 and therefore move faster while still keeping the option on the table to re-use the new v2 API for pure HTTP in the future. In contrast, re-using the HTTP API v1 for HTTP over AMQP is possible, but dirty because separate handlers (HTTP over AMQP and pure HTTP) replying differently will be needed for the same v1 endpoints.
2024-02-08 01:26:13 +08:00
rabbitmq_amqp_client:
- rabbitmq_amqp_address
Enable AMQP 1.0 clients to manage topologies ## What? * Allow AMQP 1.0 clients to dynamically create and delete RabbitMQ topologies (exchanges, queues, bindings). * Provide an Erlang AMQP 1.0 client that manages topologies. ## Why? Today, RabbitMQ topologies can be created via: * [Management HTTP API](https://www.rabbitmq.com/docs/management#http-api) (including Management UI and [messaging-topology-operator](https://github.com/rabbitmq/messaging-topology-operator)) * [Definition Import](https://www.rabbitmq.com/docs/definitions#import) * AMQP 0.9.1 clients Up to RabbitMQ 3.13 the RabbitMQ AMQP 1.0 plugin auto creates queues and bindings depending on the terminus [address format](https://github.com/rabbitmq/rabbitmq-server/tree/v3.13.x/deps/rabbitmq_amqp1_0#routing-and-addressing). Such implicit creation of topologies is limiting and obscure. For some address formats, queues will be created, but not deleted. Some of RabbitMQ's success is due to its flexible routing topologies that AMQP 0.9.1 clients can create and delete dynamically. This commit allows dynamic management of topologies for AMQP 1.0 clients. This commit builds on top of Native AMQP 1.0 (PR #9022) and will be available in RabbitMQ 4.0. ## How? This commits adds the following management operations for AMQP 1.0 clients: * declare queue * delete queue * purge queue * bind queue to exchange * unbind queue from exchange * declare exchange * delete exchange * bind exchange to exchange * unbind exchange from exchange Hence, at least the AMQP 0.9.1 management operations are supported for AMQP 1.0 clients. In addition the operation * get queue is provided which - similar to `declare queue` - returns queue information including the current leader and replicas. This allows clients to publish or consume locally on the node that hosts the queue. Compared to AMQP 0.9.1 whose commands and command fields are fixed, the new AMQP Management API is extensible: New operations and new fields can easily be added in the future. There are different design options how management operations could be supported for AMQP 1.0 clients: 1. Use a special exchange type as done in https://github.com/rabbitmq/rabbitmq-management-exchange This has the advantage that any protocol client (e.g. also STOMP clients) could dynamically manage topologies. However, a special exchange type is the wrong abstraction. 2. Clients could send "special" messages with special headers that the broker interprets. This commit decided for a variation of the 2nd option using a more standardized way by re-using a subest of the following latest AMQP 1.0 extension specifications: * [AMQP Request-Response Messaging with Link Pairing Version 1.0 - Committee Specification 01](https://docs.oasis-open.org/amqp/linkpair/v1.0/cs01/linkpair-v1.0-cs01.html) (February 2021) * [HTTP Semantics and Content over AMQP Version 1.0 - Working Draft 06](https://groups.oasis-open.org/higherlogic/ws/public/document?document_id=65571) (July 2019) * [AMQP Management Version 1.0 - Working Draft 16](https://groups.oasis-open.org/higherlogic/ws/public/document?document_id=65575) (July 2019) An important goal is to keep the interaction between AMQP 1.0 client and RabbitMQ simple to increase usage, development and adoptability of future RabbitMQ AMQP 1.0 client library wrappers. The AMQP 1.0 client has to create a link pair to the special `/management` node. This allows the client to send and receive from the management node. Similar to AMQP 0.9.1, there is no need for a reply queue since the reply will be sent directly to the client. Requests and responses are modelled via HTTP, but sent via AMQP using the `HTTP Semantics and Content over AMQP` extension (henceforth `HTTP over AMQP` extension). This commit tries to follow the `HTTP over AMQP` extension as much as possible but deviates where this draft spec doesn't make sense. The projected mode §4.1 is used as opposed to tunneled mode §4.2. A named relay `/management` is used (§6.3) where the message field `to` is the URL. Deviations are * §3.1 mandates that URIs are not encoded in an AMQP message. However, we percent encode URIs in the AMQP message. Otherwise there is for example no way to distinguish a `/` in a queue name from the URI path separator `/`. * §4.1.4 mandates a data section. This commit uses an amqp-value section as it's a better fit given that the content is AMQP encoded data. Using an HTTP API allows for a common well understood interface and future extensibility. Instead of re-using the current RabbitMQ HTTP API, this commit uses a new HTTP API (let's call it v2) which could be used as a future API for plain HTTP clients. ### HTTP API v1 The current HTTP API (let's call it v1) is **not** used since v1 comes with a couple of weaknesses: 1. Deep level of nesting becomes confusing and difficult to manage. Examples of deep nesting in v1: ``` /api/bindings/vhost/e/source/e/destination/props /api/bindings/vhost/e/exchange/q/queue/props ``` 2. Redundant endpoints returning the same resources v1 has 9 endpoints to list binding(s): ``` /api/exchanges/vhost/name/bindings/source /api/exchanges/vhost/name/bindings/destination /api/queues/vhost/name/bindings /api/bindings /api/bindings/vhost /api/bindings/vhost/e/exchange/q/queue /api/bindings/vhost/e/exchange/q/queue/props /api/bindings/vhost/e/source/e/destination /api/bindings/vhost/e/source/e/destination/props ``` 3. Verbs in path names Path names should be nouns instead. v1 contains verbs: ``` /api/queues/vhost/name/get /api/exchanges/vhost/name/publish ``` ### AMQP Management extension Only few aspects of the AMQP Management extension are used. The central idea of the AMQP management spec is **dynamic discovery** such that broker independent AMQP 1.0 clients can discover objects, types, operations, and HTTP endpoints of specific brokers. In fact, clients are only conformant if: > All request addresses are dynamically discovered starting from the discovery document. > A requesting container MUST NOT use fixed assumptions about the addressing structure of the management API. While this is a nice and powerful idea, no AMQP 1.0 client and no AMQP 1.0 server implement the latest AMQP 1.0 management spec from 2019, partly presumably due to its complexity. Therefore, the idea of such dynamic discovery has failed to be implemented in practice. The AMQP management spec mandates that the management endpoint returns a discovery document containing broker specific collections, types, configuration, and operations including their endpoints. The API endpoints of the AMQP management spec are therefore all designed around dynamic discovery. For example, to create either a queue or an exchange, the client has to ``` POST /$management/entities ``` which shows that the entities collection acts as a generic factory, see section 2.2. The server will then create the resource and reply with a location header containing a URI pointing to the resource. For RabbitMQ, we don’t need such a generic factory to create queues or exchanges. To list bindings for a queue Q1, the spec suggests ``` GET /$management/Queues/Q1/$management/entities ``` which again shows the generic entities endpoint as well as a `$management` endpoint under Q1 to allow a queue to return a discovery document. For RabbitMQ, we don’t need such generic endpoints and discovery documents. Given we aim for our own thin RabbitMQ AMQP 1.0 client wrapper libraries which expose the RabbitMQ model to the developer, we can directly use fixed HTTP endpoint assumptions in our RabbitMQ specific libraries. This is by far simpler than using the dynamic endpoints of the management spec. Simplicity leads to higher adoption and enables more developers to write RabbitMQ AMQP 1.0 client library wrappers. The AMQP Management extension also suffers from deep level of nesting in paths Examples: ``` /$management/Queues/Q1/$management/entities /$management/Queues/Q1/Bindings/Binding1 ``` as well as verbs in path names: Section 7.1.4 suggests using verbs in path names, for example “purge”, due to the dynamic operations discovery document. ### HTTP API v2 This commit introduces a new HTTP API v2 following best practices. It could serve as a future API for plain HTTP clients. This commit and RabbitMQ 4.0 will only implement a minimal set of HTTP API v2 endpoints and only for HTTP over AMQP. In other words, the existing HTTP API v1 Cowboy handlers will continue to be used for all plain HTTP requests in RabbitMQ 4.0 and will remain untouched for RabbitMQ 4.0. Over time, after 4.0 shipped, we could ship a pure HTTP API implementation for HTTP API v2. Hence, the new HTTP API v2 endpoints for HTTP over AMQP should be designed such that they can be re-used in the future for a pure HTTP implementation. The minimal set of endpoints for RabbitMQ 4.0 are: `` GET / PUT / DELETE /vhosts/:vhost/queues/:queue ``` read, create, delete a queue ``` DELETE /vhosts/:vhost/queues/:queue/messages ``` purges a queue ``` GET / DELETE /vhosts/:vhost/bindings/:binding ``` read, delete bindings where `:binding` is a binding ID of the following path segment: ``` src=e1;dstq=q2;key=my-key;args= ``` Binding arguments `args` has an empty value by default, i.e. there are no binding arguments. If the binding includes binding arguments, `args` will be an Erlang portable term hash provided by the server similar to what’s provided in HTTP API v1 today. Alternatively, we could use an arguments scheme of: ``` args=k1,utf8,v1&k2,uint,3 ``` However, such a scheme leads to long URIs when there are many binding arguments. Note that it’s perfectly fine for URI producing applications to include URI reserved characters `=` / `;` / `,` / `$` in a path segment. To create a binding, the client therefore needs to POST to a bindings factory URI: ``` POST /vhosts/:vhost/bindings ``` To list all bindings between a source exchange e1 and destination exchange e2 with binding key k1: ``` GET /vhosts/:vhost/bindings?src=e1&dste=e2&key=k1 ``` This endpoint will be called by the RabbitMQ AMQP 1.0 client library to unbind a binding with non-empty binding arguments to get the binding ID before invoking a ``` DELETE /vhosts/:vhost/bindings/:binding ``` In future, after RabbitMQ 4.0 shipped, new API endpoints could be added. The following is up for discussion and is only meant to show the clean and simple design of HTTP API v2. Bindings endpoint can be queried as follows: to list all bindings for a given source exchange e1: ``` GET /vhosts/:vhost/bindings?src=e1 ``` to list all bindings for a given destination queue q1: ``` GET /vhosts/:vhost/bindings?dstq=q1 ``` to list all bindings between a source exchange e1 and destination queue q1: ``` GET /vhosts/:vhost/bindings?src=e1&dstq=q1 ``` multiple bindings between source exchange e1 and destination queue q1 could be deleted at once as follows: ``` DELETE /vhosts/:vhost/bindings?src=e1&dstq=q1 ``` GET could be supported globally across all vhosts: ``` /exchanges /queues /bindings ``` Publish a message: ``` POST /vhosts/:vhost/queues/:queue/messages ``` Consume or peek a message (depending on query parameters): ``` GET /vhosts/:vhost/queues/:queue/messages ``` Note that the AMQP 1.0 client omits the `/vhost/:vhost` path prefix. Since an AMQP connection belongs to a single vhost, there is no need to additionally include the vhost in every HTTP request. Pros of HTTP API v2: 1. Low level of nesting Queues, exchanges, bindings are top level entities directly under vhosts. Although the HTTP API doesn’t have to reflect how resources are stored in the database, v2 does nicely reflect the Khepri tree structure. 2. Nouns instead of verbs HTTP API v2 is very simple to read and understand as shown by ``` POST /vhosts/:vhost/queues/:queue/messages to post messages, i.e. publish to a queue. GET /vhosts/:vhost/queues/:queue/messages to get messages, i.e. consume or peek from a queue. DELETE /vhosts/:vhost/queues/:queue/messages to delete messages, i.e. purge a queue. ``` A separate new HTTP API v2 allows us to ship only handlers for HTTP over AMQP for RabbitMQ 4.0 and therefore move faster while still keeping the option on the table to re-use the new v2 API for pure HTTP in the future. In contrast, re-using the HTTP API v1 for HTTP over AMQP is possible, but dirty because separate handlers (HTTP over AMQP and pure HTTP) replying differently will be needed for the same v1 endpoints.
2024-02-08 01:26:13 +08:00
- rabbitmq_amqp_client
rabbitmq_amqp1_0:
Support AMQP 1.0 natively ## What Similar to Native MQTT in #5895, this commits implements Native AMQP 1.0. By "native", we mean do not proxy via AMQP 0.9.1 anymore. ## Why Native AMQP 1.0 comes with the following major benefits: 1. Similar to Native MQTT, this commit provides better throughput, latency, scalability, and resource usage for AMQP 1.0. See https://blog.rabbitmq.com/posts/2023/03/native-mqtt for native MQTT improvements. See further below for some benchmarks. 2. Since AMQP 1.0 is not limited anymore by the AMQP 0.9.1 protocol, this commit allows implementing more AMQP 1.0 features in the future. Some features are already implemented in this commit (see next section). 3. Simpler, better understandable, and more maintainable code. Native AMQP 1.0 as implemented in this commit has the following major benefits compared to AMQP 0.9.1: 4. Memory and disk alarms will only stop accepting incoming TRANSFER frames. New connections can still be created to consume from RabbitMQ to empty queues. 5. Due to 4. no need anymore for separate connections for publishers and consumers as we currently recommended for AMQP 0.9.1. which potentially halves the number of physical TCP connections. 6. When a single connection sends to multiple target queues, a single slow target queue won't block the entire connection. Publisher can still send data quickly to all other target queues. 7. A publisher can request whether it wants publisher confirmation on a per-message basis. In AMQP 0.9.1 publisher confirms are configured per channel only. 8. Consumers can change their "prefetch count" dynamically which isn't possible in our AMQP 0.9.1 implementation. See #10174 9. AMQP 1.0 is an extensible protocol This commit also fixes dozens of bugs present in the AMQP 1.0 plugin in RabbitMQ 3.x - most of which cannot be backported due to the complexity and limitations of the old 3.x implementation. This commit contains breaking changes and is therefore targeted for RabbitMQ 4.0. ## Implementation details 1. Breaking change: With Native AMQP, the behaviour of ``` Convert AMQP 0.9.1 message headers to application properties for an AMQP 1.0 consumer amqp1_0.convert_amqp091_headers_to_app_props = false | true (default false) Convert AMQP 1.0 Application Properties to AMQP 0.9.1 headers amqp1_0.convert_app_props_to_amqp091_headers = false | true (default false) ``` will break because we always convert according to the message container conversions. For example, AMQP 0.9.1 x-headers will go into message-annotations instead of application properties. Also, `false` won’t be respected since we always convert the headers with message containers. 2. Remove rabbit_queue_collector rabbit_queue_collector is responsible for synchronously deleting exclusive queues. Since the AMQP 1.0 plugin never creates exclusive queues, rabbit_queue_collector doesn't need to be started in the first place. This will save 1 Erlang process per AMQP 1.0 connection. 3. 7 processes per connection + 1 process per session in this commit instead of 7 processes per connection + 15 processes per session in 3.x Supervision hierarchy got re-designed. 4. Use 1 writer process per AMQP 1.0 connection AMQP 0.9.1 uses a separate rabbit_writer Erlang process per AMQP 0.9.1 channel. Prior to this commit, AMQP 1.0 used a separate rabbit_amqp1_0_writer process per AMQP 1.0 session. Advantage of single writer proc per session (prior to this commit): * High parallelism for serialising packets if multiple sessions within a connection write heavily at the same time. This commit uses a single writer process per AMQP 1.0 connection that is shared across all AMQP 1.0 sessions. Advantages of single writer proc per connection (this commit): * Lower memory usage with hundreds of thousands of AMQP 1.0 sessions * Less TCP and IP header overhead given that the single writer process can accumulate across all sessions bytes before flushing the socket. In other words, this commit decides that a reader / writer process pair per AMQP 1.0 connection is good enough for bi-directional TRANSFER flows. Having a writer per session is too heavy. We still ensure high throughput by having separate reader, writer, and session processes. 5. Transform rabbit_amqp1_0_writer into gen_server Why: Prior to this commit, when clicking on the AMQP 1.0 writer process in observer, the process crashed. Instead of handling all these debug messages of the sys module, it's better to implement a gen_server. There is no advantage of using a special OTP process over gen_server for the AMQP 1.0 writer. gen_server also provides cleaner format status output. How: Message callbacks return a timeout of 0. After all messages in the inbox are processed, the timeout message is handled by flushing any pending bytes. 6. Remove stats timer from writer AMQP 1.0 connections haven't emitted any stats previously. 7. When there are contiguous queue confirmations in the session process mailbox, batch them. When the confirmations are sent to the publisher, a single DISPOSITION frame is sent for contiguously confirmed delivery IDs. This approach should be good enough. However it's sub optimal in scenarios where contiguous delivery IDs that need confirmations are rare, for example: * There are multiple links in the session with different sender settlement modes and sender publishes across these links interleaved. * sender settlement mode is mixed and sender publishes interleaved settled and unsettled TRANSFERs. 8. Introduce credit API v2 Why: The AMQP 0.9.1 credit extension which is to be removed in 4.0 was poorly designed since basic.credit is a synchronous call into the queue process blocking the entire AMQP 1.0 session process. How: Change the interactions between queue clients and queue server implementations: * Clients only request a credit reply if the FLOW's `echo` field is set * Include all link flow control state held by the queue process into a new credit_reply queue event: * `available` after the queue sends any deliveries * `link-credit` after the queue sends any deliveries * `drain` which allows us to combine the old queue events send_credit_reply and send_drained into a single new queue event credit_reply. * Include the consumer tag into the credit_reply queue event such that the AMQP 1.0 session process can process any credit replies asynchronously. Link flow control state `delivery-count` also moves to the queue processes. The new interactions are hidden behind feature flag credit_api_v2 to allow for rolling upgrades from 3.13 to 4.0. 9. Use serial number arithmetic in quorum queues and session process. 10. Completely bypass the rabbit_limiter module for AMQP 1.0 flow control. The goal is to eventually remove the rabbit_limiter module in 4.0 since AMQP 0.9.1 global QoS will be unsupported in 4.0. This commit lifts the AMQP 1.0 link flow control logic out of rabbit_limiter into rabbit_queue_consumers. 11. Fix credit bug for streams: AMQP 1.0 settlements shouldn't top up link credit, only FLOW frames should top up link credit. 12. Allow sender settle mode unsettled for streams since AMQP 1.0 acknowledgements to streams are no-ops (currently). 13. Fix AMQP 1.0 client bugs Auto renewing credits should not be related to settling TRANSFERs. Remove field link_credit_unsettled as it was wrong and confusing. Prior to this commit auto renewal did not work when the sender uses sender settlement mode settled. 14. Fix AMQP 1.0 client bugs The wrong outdated Link was passed to function auto_flow/2 15. Use osiris chunk iterator Only hold messages of uncompressed sub batches in memory if consumer doesn't have sufficient credits. Compressed sub batches are skipped for non Stream protocol consumers. 16. Fix incoming link flow control Always use confirms between AMQP 1.0 queue clients and queue servers. As already done internally by rabbit_fifo_client and rabbit_stream_queue, use confirms for classic queues as well. 17. Include link handle into correlation when publishing messages to target queues such that session process can correlate confirms from target queues to incoming links. 18. Only grant more credits to publishers if publisher hasn't sufficient credits anymore and there are not too many unconfirmed messages on the link. 19. Completely ignore `block` and `unblock` queue actions and RabbitMQ credit flow between classic queue process and session process. 20. Link flow control is independent between links. A client can refer to a queue or to an exchange with multiple dynamically added target queues. Multiple incoming links can also fan in to the same queue. However the link topology looks like, this commit ensures that each link is only granted more credits if that link isn't overloaded. 21. A connection or a session can send to many different queues. In AMQP 0.9.1, a single slow queue will lead to the entire channel, and then entire connection being blocked. This commit makes sure that a single slow queue from one link won't slow down sending on other links. For example, having link A sending to a local classic queue and link B sending to 5 replica quorum queue, link B will naturally grant credits slower than link A. So, despite the quorum queue being slower in confirming messages, the same AMQP 1.0 connection and session can still pump data very fast into the classic queue. 22. If cluster wide memory or disk alarm occurs. Each session sends a FLOW with incoming-window to 0 to sending client. If sending clients don’t obey, force disconnect the client. If cluster wide memory alarm clears: Each session resumes with a FLOW defaulting to initial incoming-window. 23. All operations apart of publishing TRANSFERS to RabbitMQ can continue during cluster wide alarms, specifically, attaching consumers and consuming, i.e. emptying queues. There is no need for separate AMQP 1.0 connections for publishers and consumers as recommended in our AMQP 0.9.1 implementation. 24. Flow control summary: * If queue becomes bottleneck, that’s solved by slowing down individual sending links (AMQP 1.0 link flow control). * If session becomes bottleneck (more unlikely), that’s solved by AMQP 1.0 session flow control. * If connection becomes bottleneck, it naturally won’t read fast enough from the socket causing TCP backpressure being applied. Nowhere will RabbitMQ internal credit based flow control (i.e. module credit_flow) be used on the incoming AMQP 1.0 message path. 25. Register AMQP sessions Prefer local-only pg over our custom pg_local implementation as pg is a better process group implementation than pg_local. pg_local was identified as bottleneck in tests where many MQTT clients were disconnected at once. 26. Start a local-only pg when Rabbit boots: > A scope can be kept local-only by using a scope name that is unique cluster-wide, e.g. the node name: > pg:start_link(node()). Register AMQP 1.0 connections and sessions with pg. In future we should remove pg_local and instead use the new local-only pg for all registered processes such as AMQP 0.9.1 connections and channels. 27. Requeue messages if link detached Although the spec allows to settle delivery IDs on detached links, RabbitMQ does not respect the 'closed' field of the DETACH frame and therefore handles every DETACH frame as closed. Since the link is closed, we expect every outstanding delivery to be requeued. In addition to consumer cancellation, detaching a link therefore causes in flight deliveries to be requeued. Note that this behaviour is different from merely consumer cancellation in AMQP 0.9.1: "After a consumer is cancelled there will be no future deliveries dispatched to it. Note that there can still be "in flight" deliveries dispatched previously. Cancelling a consumer will neither discard nor requeue them." [https://www.rabbitmq.com/consumers.html#unsubscribing] An AMQP receiver can first drain, and then detach to prevent "in flight" deliveries 28. Init AMQP session with BEGIN frame Similar to how there can't be an MQTT processor without a CONNECT frame, there can't be an AMQP session without a BEGIN frame. This allows having strict dialyzer types for session flow control fields (i.e. not allowing 'undefined'). 29. Move serial_number to AMQP 1.0 common lib such that it can be used by both AMQP 1.0 server and client 30. Fix AMQP client to do serial number arithmetic. 31. AMQP client: Differentiate between delivery-id and transfer-id for better understandability. 32. Fix link flow control in classic queues This commit fixes ``` java -jar target/perf-test.jar -ad false -f persistent -u cq -c 3000 -C 1000000 -y 0 ``` followed by ``` ./omq -x 0 amqp -T /queue/cq -D 1000000 --amqp-consumer-credits 2 ``` Prior to this commit, (and on RabbitMQ 3.x) the consuming would halt after around 8 - 10,000 messages. The bug was that in flight messages from classic queue process to session process were not taken into account when topping up credit to the classic queue process. Fixes #2597 The solution to this bug (and a much cleaner design anyway independent of this bug) is that queues should hold all link flow control state including the delivery-count. Hence, when credit API v2 is used the delivery-count will be held by the classic queue process, quorum queue process, and stream queue client instead of managing the delivery-count in the session. 33. The double level crediting between (a) session process and rabbit_fifo_client, and (b) rabbit_fifo_client and rabbit_fifo was removed. Therefore, instead of managing 3 separate delivery-counts (i. session, ii. rabbit_fifo_client, iii. rabbit_fifo), only 1 delivery-count is used in rabbit_fifo. This is a big simplification. 34. This commit fixes quorum queues without bumping the machine version nor introducing new rabbit_fifo commands. Whether credit API v2 is used is solely determined at link attachment time depending on whether feature flag credit_api_v2 is enabled. Even when that feature flag will be enabled later on, this link will keep using credit API v1 until detached (or the node is shut down). Eventually, after feature flag credit_api_v2 has been enabled and a subsequent rolling upgrade, all links will use credit API v2. This approach is safe and simple. The 2 alternatives to move delivery-count from the session process to the queue processes would have been: i. Explicit feature flag credit_api_v2 migration function * Can use a gen_server:call and only finish migration once all delivery-counts were migrated. Cons: * Extra new message format just for migration is required. * Risky as migration will fail if a target queue doesn’t reply. ii. Session always includes DeliveryCountSnd when crediting to the queue: Cons: * 2 delivery counts will be hold simultaneously in session proc and queue proc; could be solved by deleting the session proc’s delivery-count for credit-reply * What happens if the receiver doesn’t provide credit for a very long time? Is that a problem? 35. Support stream filtering in AMQP 1.0 (by @acogoluegnes) Use the x-stream-filter-value message annotation to carry the filter value in a published message. Use the rabbitmq:stream-filter and rabbitmq:stream-match-unfiltered filters when creating a receiver that wants to filter out messages from a stream. 36. Remove credit extension from AMQP 0.9.1 client 37. Support maintenance mode closing AMQP 1.0 connections. 38. Remove AMQP 0.9.1 client dependency from AMQP 1.0 implementation. 39. Move AMQP 1.0 plugin to the core. AMQP 1.0 is enabled by default. The old rabbitmq_amqp1_0 plugin will be kept as a no-op plugin to prevent deployment tools from failing that execute: ``` rabbitmq-plugins enable rabbitmq_amqp1_0 rabbitmq-plugins disable rabbitmq_amqp1_0 ``` 40. Breaking change: Remove CLI command `rabbitmqctl list_amqp10_connections`. Instead, list both AMQP 0.9.1 and AMQP 1.0 connections in `list_connections`: ``` rabbitmqctl list_connections protocol Listing connections ... protocol {1, 0} {0,9,1} ``` ## Benchmarks ### Throughput & Latency Setup: * Single node Ubuntu 22.04 * Erlang 26.1.1 Start RabbitMQ: ``` make run-broker PLUGINS="rabbitmq_management rabbitmq_amqp1_0" FULL=1 RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+S 3" ``` Predeclare durable classic queue cq1, durable quorum queue qq1, durable stream queue sq1. Start client: https://github.com/ssorj/quiver https://hub.docker.com/r/ssorj/quiver/tags (digest 453a2aceda64) ``` docker run -it --rm --add-host host.docker.internal:host-gateway ssorj/quiver:latest bash-5.1# quiver --version quiver 0.4.0-SNAPSHOT ``` 1. Classic queue ``` quiver //host.docker.internal//amq/queue/cq1 --durable --count 1m --duration 10m --body-size 12 --credit 1000 ``` This commit: ``` Count ............................................. 1,000,000 messages Duration ............................................... 73.8 seconds Sender rate .......................................... 13,548 messages/s Receiver rate ........................................ 13,547 messages/s End-to-end rate ...................................... 13,547 messages/s Latencies by percentile: 0% ........ 0 ms 90.00% ........ 9 ms 25% ........ 2 ms 99.00% ....... 14 ms 50% ........ 4 ms 99.90% ....... 17 ms 100% ....... 26 ms 99.99% ....... 24 ms ``` RabbitMQ 3.x (main branch as of 30 January 2024): ``` ---------------------- Sender ----------------------- --------------------- Receiver ---------------------- -------- Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Lat [ms] ----------------------------------------------------- ----------------------------------------------------- -------- 2.1 130,814 65,342 6 73.6 2.1 3,217 1,607 0 8.0 511 4.1 163,580 16,367 2 74.1 4.1 3,217 0 0 8.0 0 6.1 229,114 32,767 3 74.1 6.1 3,217 0 0 8.0 0 8.1 261,880 16,367 2 74.1 8.1 67,874 32,296 8 8.2 7,662 10.1 294,646 16,367 2 74.1 10.1 67,874 0 0 8.2 0 12.1 360,180 32,734 3 74.1 12.1 67,874 0 0 8.2 0 14.1 392,946 16,367 3 74.1 14.1 68,604 365 0 8.2 12,147 16.1 458,480 32,734 3 74.1 16.1 68,604 0 0 8.2 0 18.1 491,246 16,367 2 74.1 18.1 68,604 0 0 8.2 0 20.1 556,780 32,767 4 74.1 20.1 68,604 0 0 8.2 0 22.1 589,546 16,375 2 74.1 22.1 68,604 0 0 8.2 0 receiver timed out 24.1 622,312 16,367 2 74.1 24.1 68,604 0 0 8.2 0 quiver: error: PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/cq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-otujr23y' returned non-zero exit status 1. Traceback (most recent call last): File "/usr/local/lib/quiver/python/quiver/pair.py", line 144, in run _plano.wait(receiver, check=True) File "/usr/local/lib/quiver/python/plano/main.py", line 1243, in wait raise PlanoProcessError(proc) plano.main.PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/cq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-otujr23y' returned non-zero exit status 1. ``` 2. Quorum queue: ``` quiver //host.docker.internal//amq/queue/qq1 --durable --count 1m --duration 10m --body-size 12 --credit 1000 ``` This commit: ``` Count ............................................. 1,000,000 messages Duration .............................................. 101.4 seconds Sender rate ........................................... 9,867 messages/s Receiver rate ......................................... 9,868 messages/s End-to-end rate ....................................... 9,865 messages/s Latencies by percentile: 0% ....... 11 ms 90.00% ....... 23 ms 25% ....... 15 ms 99.00% ....... 28 ms 50% ....... 18 ms 99.90% ....... 33 ms 100% ....... 49 ms 99.99% ....... 47 ms ``` RabbitMQ 3.x: ``` ---------------------- Sender ----------------------- --------------------- Receiver ---------------------- -------- Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Time [s] Count [m] Rate [m/s] CPU [%] RSS [M] Lat [ms] ----------------------------------------------------- ----------------------------------------------------- -------- 2.1 130,814 65,342 9 69.9 2.1 18,430 9,206 5 7.6 1,221 4.1 163,580 16,375 5 70.2 4.1 18,867 218 0 7.6 2,168 6.1 229,114 32,767 6 70.2 6.1 18,867 0 0 7.6 0 8.1 294,648 32,734 7 70.2 8.1 18,867 0 0 7.6 0 10.1 360,182 32,734 6 70.2 10.1 18,867 0 0 7.6 0 12.1 425,716 32,767 6 70.2 12.1 18,867 0 0 7.6 0 receiver timed out 14.1 458,482 16,367 5 70.2 14.1 18,867 0 0 7.6 0 quiver: error: PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/qq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-b1gcup43' returned non-zero exit status 1. Traceback (most recent call last): File "/usr/local/lib/quiver/python/quiver/pair.py", line 144, in run _plano.wait(receiver, check=True) File "/usr/local/lib/quiver/python/plano/main.py", line 1243, in wait raise PlanoProcessError(proc) plano.main.PlanoProcessError: Command 'quiver-arrow receive //host.docker.internal//amq/queue/qq1 --impl qpid-proton-c --duration 10m --count 1m --rate 0 --body-size 12 --credit 1000 --transaction-size 0 --timeout 10 --durable --output /tmp/quiver-b1gcup43' returned non-zero exit status 1. ``` 3. Stream: ``` quiver-arrow send //host.docker.internal//amq/queue/sq1 --durable --count 1m -d 10m --summary --verbose ``` This commit: ``` Count ............................................. 1,000,000 messages Duration ................................................ 8.7 seconds Message rate ........................................ 115,154 messages/s ``` RabbitMQ 3.x: ``` Count ............................................. 1,000,000 messages Duration ............................................... 21.2 seconds Message rate ......................................... 47,232 messages/s ``` ### Memory usage Start RabbitMQ: ``` ERL_MAX_PORTS=3000000 RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+P 3000000 +S 6" make run-broker PLUGINS="rabbitmq_amqp1_0" FULL=1 RABBITMQ_CONFIG_FILE="rabbitmq.conf" ``` ``` /bin/cat rabbitmq.conf tcp_listen_options.sndbuf = 2048 tcp_listen_options.recbuf = 2048 vm_memory_high_watermark.relative = 0.95 vm_memory_high_watermark_paging_ratio = 0.95 loopback_users = none ``` Create 50k connections with 2 sessions per connection, i.e. 100k session in total: ```go package main import ( "context" "log" "time" "github.com/Azure/go-amqp" ) func main() { for i := 0; i < 50000; i++ { conn, err := amqp.Dial(context.TODO(), "amqp://nuc", &amqp.ConnOptions{SASLType: amqp.SASLTypeAnonymous()}) if err != nil { log.Fatal("dialing AMQP server:", err) } _, err = conn.NewSession(context.TODO(), nil) if err != nil { log.Fatal("creating AMQP session:", err) } _, err = conn.NewSession(context.TODO(), nil) if err != nil { log.Fatal("creating AMQP session:", err) } } log.Println("opened all connections") time.Sleep(5 * time.Hour) } ``` This commit: ``` erlang:memory(). [{total,4586376480}, {processes,4025898504}, {processes_used,4025871040}, {system,560477976}, {atom,1048841}, {atom_used,1042841}, {binary,233228608}, {code,21449982}, {ets,108560464}] erlang:system_info(process_count). 450289 ``` 7 procs per connection + 1 proc per session. (7 + 2*1) * 50,000 = 450,000 procs RabbitMQ 3.x: ``` erlang:memory(). [{total,15168232704}, {processes,14044779256}, {processes_used,14044755120}, {system,1123453448}, {atom,1057033}, {atom_used,1052587}, {binary,236381264}, {code,21790238}, {ets,391423744}] erlang:system_info(process_count). 1850309 ``` 7 procs per connection + 15 per session (7 + 2*15) * 50,000 = 1,850,000 procs 50k connections + 100k session require with this commit: 4.5 GB in RabbitMQ 3.x: 15 GB ## Future work 1. More efficient parser and serializer 2. TODO in mc_amqp: Do not store the parsed message on disk. 3. Implement both AMQP HTTP extension and AMQP management extension to allow AMQP clients to create RabbitMQ objects (queues, exchanges, ...).
2023-07-21 18:29:07 +08:00
- rabbitmq_amqp1_0_noop
rabbitmq_auth_backend_cache:
- rabbit_auth_backend_cache
- rabbit_auth_backend_cache_app
- rabbit_auth_cache
- rabbit_auth_cache_dict
- rabbit_auth_cache_ets
- rabbit_auth_cache_ets_segmented
- rabbit_auth_cache_ets_segmented_stateless
rabbitmq_auth_backend_http:
- rabbit_auth_backend_http
- rabbit_auth_backend_http_app
rabbitmq_auth_backend_ldap:
- rabbit_auth_backend_ldap
- rabbit_auth_backend_ldap_app
- rabbit_auth_backend_ldap_util
- rabbit_log_ldap
rabbitmq_auth_backend_oauth2:
- Elixir.RabbitMQ.CLI.Ctl.Commands.AddSigningKeyCommand
- Elixir.RabbitMQ.CLI.Ctl.Commands.AddUaaKeyCommand
- rabbit_auth_backend_oauth2
- rabbit_auth_backend_oauth2_app
2024-01-03 16:28:36 +08:00
- rabbit_oauth2_config
- rabbit_oauth2_scope
- uaa_jwks
- uaa_jwt
- uaa_jwt_jwk
- uaa_jwt_jwt
- wildcard
rabbitmq_auth_mechanism_ssl:
- rabbit_auth_mechanism_ssl
- rabbit_auth_mechanism_ssl_app
rabbitmq_aws:
- rabbitmq_aws
- rabbitmq_aws_app
- rabbitmq_aws_config
- rabbitmq_aws_json
- rabbitmq_aws_sign
- rabbitmq_aws_sup
- rabbitmq_aws_urilib
- rabbitmq_aws_xml
rabbitmq_consistent_hash_exchange:
- Elixir.RabbitMQ.CLI.Diagnostics.Commands.ConsistentHashExchangeRingStateCommand
- rabbit_db_ch_exchange
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
- rabbit_db_ch_exchange_m2k_converter
- rabbit_exchange_type_consistent_hash
rabbitmq_ct_client_helpers:
- rabbit_ct_client_helpers
rabbitmq_ct_helpers:
- cth_log_redirect_any_domains
- rabbit_control_helper
- rabbit_ct_broker_helpers
- rabbit_ct_config_schema
- rabbit_ct_helpers
- rabbit_ct_proper_helpers
- rabbit_ct_vm_helpers
- rabbit_mgmt_test_util
rabbitmq_event_exchange:
- rabbit_event_exchange_decorator
- rabbit_exchange_type_event
rabbitmq_federation:
- Elixir.RabbitMQ.CLI.Ctl.Commands.FederationStatusCommand
- Elixir.RabbitMQ.CLI.Ctl.Commands.RestartFederationLinkCommand
- rabbit_federation_app
- rabbit_federation_db
- rabbit_federation_event
- rabbit_federation_exchange
- rabbit_federation_exchange_link
- rabbit_federation_exchange_link_sup_sup
- rabbit_federation_link_sup
- rabbit_federation_link_util
- rabbit_federation_parameters
- rabbit_federation_pg
- rabbit_federation_queue
- rabbit_federation_queue_link
- rabbit_federation_queue_link_sup_sup
- rabbit_federation_status
- rabbit_federation_sup
- rabbit_federation_upstream
- rabbit_federation_upstream_exchange
- rabbit_federation_util
- rabbit_log_federation
rabbitmq_federation_management:
- rabbit_federation_mgmt
rabbitmq_federation_prometheus:
- rabbit_federation_prometheus_app
- rabbit_federation_prometheus_collector
- rabbit_federation_prometheus_sup
rabbitmq_jms_topic_exchange:
- rabbit_db_jms_exchange
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
- rabbit_db_jms_exchange_m2k_converter
- rabbit_jms_topic_exchange
- sjx_evaluator
rabbitmq_management:
- rabbit_mgmt_app
- rabbit_mgmt_cors
- rabbit_mgmt_csp
- rabbit_mgmt_db
- rabbit_mgmt_db_cache
- rabbit_mgmt_db_cache_sup
- rabbit_mgmt_dispatcher
- rabbit_mgmt_extension
- rabbit_mgmt_features
- rabbit_mgmt_headers
- rabbit_mgmt_hsts
- rabbit_mgmt_load_definitions
- rabbit_mgmt_login
- rabbit_mgmt_nodes
- rabbit_mgmt_oauth_bootstrap
- rabbit_mgmt_reset_handler
- rabbit_mgmt_stats
- rabbit_mgmt_sup
- rabbit_mgmt_sup_sup
- rabbit_mgmt_util
- rabbit_mgmt_wm_aliveness_test
- rabbit_mgmt_wm_auth
- rabbit_mgmt_wm_auth_attempts
- rabbit_mgmt_wm_binding
- rabbit_mgmt_wm_bindings
- rabbit_mgmt_wm_channel
- rabbit_mgmt_wm_channels
- rabbit_mgmt_wm_channels_vhost
- rabbit_mgmt_wm_cluster_name
- rabbit_mgmt_wm_connection
- rabbit_mgmt_wm_connection_channels
- rabbit_mgmt_wm_connection_user_name
- rabbit_mgmt_wm_connections
- rabbit_mgmt_wm_connections_vhost
- rabbit_mgmt_wm_consumers
- rabbit_mgmt_wm_definitions
2023-12-14 12:35:35 +08:00
- rabbit_mgmt_wm_deprecated_features
- rabbit_mgmt_wm_environment
- rabbit_mgmt_wm_exchange
- rabbit_mgmt_wm_exchange_publish
- rabbit_mgmt_wm_exchanges
- rabbit_mgmt_wm_extensions
- rabbit_mgmt_wm_feature_flag_enable
- rabbit_mgmt_wm_feature_flags
- rabbit_mgmt_wm_global_parameter
- rabbit_mgmt_wm_global_parameters
- rabbit_mgmt_wm_hash_password
- rabbit_mgmt_wm_health_check_alarms
- rabbit_mgmt_wm_health_check_certificate_expiration
- rabbit_mgmt_wm_health_check_local_alarms
- rabbit_mgmt_wm_health_check_node_is_quorum_critical
- rabbit_mgmt_wm_health_check_port_listener
- rabbit_mgmt_wm_health_check_protocol_listener
- rabbit_mgmt_wm_health_check_virtual_hosts
- rabbit_mgmt_wm_healthchecks
- rabbit_mgmt_wm_limit
- rabbit_mgmt_wm_limits
- rabbit_mgmt_wm_login
- rabbit_mgmt_wm_node
- rabbit_mgmt_wm_node_memory
- rabbit_mgmt_wm_node_memory_ets
- rabbit_mgmt_wm_nodes
- rabbit_mgmt_wm_operator_policies
- rabbit_mgmt_wm_operator_policy
- rabbit_mgmt_wm_overview
- rabbit_mgmt_wm_parameter
- rabbit_mgmt_wm_parameters
- rabbit_mgmt_wm_permission
- rabbit_mgmt_wm_permissions
- rabbit_mgmt_wm_permissions_user
- rabbit_mgmt_wm_permissions_vhost
- rabbit_mgmt_wm_policies
- rabbit_mgmt_wm_policy
- rabbit_mgmt_wm_queue
- rabbit_mgmt_wm_queue_actions
- rabbit_mgmt_wm_queue_get
- rabbit_mgmt_wm_queue_purge
- rabbit_mgmt_wm_queues
- rabbit_mgmt_wm_quorum_queue_replicas_add_member
- rabbit_mgmt_wm_quorum_queue_replicas_delete_member
- rabbit_mgmt_wm_quorum_queue_replicas_grow
- rabbit_mgmt_wm_quorum_queue_replicas_shrink
2024-08-13 03:10:32 +08:00
- rabbit_mgmt_wm_quorum_queue_status
- rabbit_mgmt_wm_rebalance_queues
- rabbit_mgmt_wm_redirect
- rabbit_mgmt_wm_reset
- rabbit_mgmt_wm_static
- rabbit_mgmt_wm_topic_permission
- rabbit_mgmt_wm_topic_permissions
- rabbit_mgmt_wm_topic_permissions_user
- rabbit_mgmt_wm_topic_permissions_vhost
- rabbit_mgmt_wm_user
- rabbit_mgmt_wm_user_limit
- rabbit_mgmt_wm_user_limits
- rabbit_mgmt_wm_users
- rabbit_mgmt_wm_users_bulk_delete
- rabbit_mgmt_wm_vhost
- rabbit_mgmt_wm_vhost_restart
- rabbit_mgmt_wm_vhosts
- rabbit_mgmt_wm_whoami
rabbitmq_management_agent:
- Elixir.RabbitMQ.CLI.Ctl.Commands.ResetStatsDbCommand
- exometer_slide
- rabbit_mgmt_agent_app
- rabbit_mgmt_agent_config
- rabbit_mgmt_agent_sup
- rabbit_mgmt_agent_sup_sup
- rabbit_mgmt_data
- rabbit_mgmt_data_compat
- rabbit_mgmt_db_handler
- rabbit_mgmt_external_stats
- rabbit_mgmt_ff
- rabbit_mgmt_format
- rabbit_mgmt_gc
- rabbit_mgmt_metrics_collector
- rabbit_mgmt_metrics_gc
- rabbit_mgmt_storage
rabbitmq_mqtt:
- Elixir.RabbitMQ.CLI.Ctl.Commands.ListMqttConnectionsCommand
Message Containers (#5077) This PR implements an approach for a "protocol (data format) agnostic core" where the format of the message isn't converted at point of reception. Currently all non AMQP 0.9.1 originating messages are converted into a AMQP 0.9.1 flavoured basic_message record before sent to a queue. If the messages are then consumed by the originating protocol they are converted back from AMQP 0.9.1. For some protocols such as MQTT 3.1 this isn't too expensive as MQTT is mostly a fairly easily mapped subset of AMQP 0.9.1 but for others such as AMQP 1.0 the conversions are awkward and in some cases lossy even if consuming from the originating protocol. This PR instead wraps all incoming messages in their originating form into a generic, extensible message container type (mc). The container module exposes an API to get common message details such as size and various properties (ttl, priority etc) directly from the source data type. Each protocol needs to implement the mc behaviour such that when a message originating form one protocol is consumed by another protocol we convert it to the target protocol at that point. The message container also contains annotations, dead letter records and other meta data we need to record during the lifetime of a message. The original protocol message is never modified unless it is consumed. This includes conversion modules to and from amqp, amqpl (AMQP 0.9.1) and mqtt. COMMIT HISTORY: * Refactor away from using the delivery{} record In many places including exchange types. This should make it easier to move towards using a message container type instead of basic_message. Add mc module and move direct replies outside of exchange Lots of changes incl classic queues Implement stream support incl amqp conversions simplify mc state record move mc.erl mc dlx stuff recent history exchange Make tracking work But doesn't take a protocol agnostic approach as we just convert everything into AMQP legacy and back. Might be good enough for now. Tracing as a whole may want a bit of a re-vamp at some point. tidy make quorum queue peek work by legacy conversion dead lettering fixes dead lettering fixes CMQ fixes rabbit_trace type fixes fixes fix Fix classic queue props test assertion fix feature flag and backwards compat Enable message_container feature flag in some SUITEs Dialyzer fixes fixes fix test fixes Various Manually update a gazelle generated file until a gazelle enhancement can be made https://github.com/rabbitmq/rules_erlang/issues/185 Add message_containers_SUITE to bazel and regen bazel files with gazelle from rules_erlang@main Simplify essential proprty access Such as durable, ttl and priority by extracting them into annotations at message container init time. Move type to remove dependenc on amqp10 stuff in mc.erl mostly because I don't know how to make bazel do the right thing add more stuff Refine routing header stuff wip Cosmetics Do not use "maybe" as type name as "maybe" is a keyword since OTP 25 which makes Erlang LS complain. * Dedup death queue names * Fix function clause crashes Fix failing tests in the MQTT shared_SUITE: A classic queue message ID can be undefined as set in https://github.com/rabbitmq/rabbitmq-server/blob/fbe79ff47b4edbc0fd95457e623d6593161ad198/deps/rabbit/src/rabbit_classic_queue_index_v2.erl#L1048 Fix failing tests in the MQTT shared_SUITE-mixed: When feature flag message_containers is disabled, the message is not an #mc{} record, but a #basic_message{} record. * Fix is_utf8_no_null crash Prior to this commit, the function crashed if invalid UTF-8 was provided, e.g.: ``` 1> rabbit_misc:is_valid_shortstr(<<"😇"/utf16>>). ** exception error: no function clause matching rabbit_misc:is_utf8_no_null(<<216,61,222,7>>) (rabbit_misc.erl, line 1481) ``` * Implement mqtt mc behaviour For now via amqp translation. This is still work in progress, but the following SUITEs pass: ``` make -C deps/rabbitmq_mqtt ct-shared t=[mqtt,v5,cluster_size_1] FULL=1 make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_1] FULL=1 ``` * Shorten mc file names Module name length matters because for each persistent message the #mc{} record is persisted to disk. ``` 1> iolist_size(term_to_iovec({mc, rabbit_mc_amqp_legacy})). 30 2> iolist_size(term_to_iovec({mc, mc_amqpl})). 17 ``` This commit renames the mc modules: ``` ag -l rabbit_mc_amqp_legacy | xargs sed -i 's/rabbit_mc_amqp_legacy/mc_amqpl/g' ag -l rabbit_mc_amqp | xargs sed -i 's/rabbit_mc_amqp/mc_amqp/g' ag -l rabbit_mqtt_mc | xargs sed -i 's/rabbit_mqtt_mc/mc_mqtt/g' ``` * mc: make deaths an annotation + fixes * Fix mc_mqtt protocol_state callback * Fix test will_delay_node_restart ``` make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_3]:will_delay_node_restart FULL=1 ``` * Bazel run gazelle * mix format rabbitmqctl.ex * Ensure ttl annotation is refelected in amqp legacy protocol state * Fix id access in message store * Fix rabbit_message_interceptor_SUITE * dializer fixes * Fix rabbit:rabbit_message_interceptor_SUITE-mixed set_annotation/3 should not result in duplicate keys * Fix MQTT shared_SUITE-mixed Up to 3.12 non-MQTT publishes were always QoS 1 regardless of delivery_mode. https://github.com/rabbitmq/rabbitmq-server/blob/75a953ce286a10aca910c098805a4f545989af38/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl#L2075-L2076 From now on, non-MQTT publishes are QoS 1 if durable. This makes more sense. The MQTT plugin must send a #basic_message{} to an old node that does not understand message containers. * Field content of 'v1_0.data' can be binary Fix ``` bazel test //deps/rabbitmq_mqtt:shared_SUITE-mixed \ --test_env FOCUS="-group [mqtt,v4,cluster_size_1] -case trace" \ -t- --test_sharding_strategy=disabled ``` * Remove route/2 and implement route/3 for all exchange types. This removes the route/2 callback from rabbit_exchange_type and makes route/3 mandatory instead. This is a breaking change and will require all implementations of exchange types to update their code, however this is necessary anyway for them to correctly handle the mc type. stream filtering fixes * Translate directly from MQTT to AMQP 0.9.1 * handle undecoded properties in mc_compat amqpl: put clause in right order recover death deatails from amqp data * Replace callback init_amqp with convert_from * Fix return value of lists:keyfind/3 * Translate directly from AMQP 0.9.1 to MQTT * Fix MQTT payload size MQTT payload can be a list when converted from AMQP 0.9.1 for example First conversions tests Plus some other conversion related fixes. bazel bazel translate amqp 1.0 null to undefined mc: property/2 and correlation_id/message_id return type tagged values. To ensure we can support a variety of types better. The type type tags are AMQP 1.0 flavoured. fix death recovery mc_mqtt: impl new api Add callbacks to allow protocols to compact data before storage And make readable if needing to query things repeatedly. bazel fix * more decoding * tracking mixed versions compat * mc: flip default of `durable` annotation to save some data. Assuming most messages are durable and that in memory messages suffer less from persistence overhead it makes sense for a non existent `durable` annotation to mean durable=true. * mc conversion tests and tidy up * mc make x_header unstrict again * amqpl: death record fixes * bazel * amqp -> amqpl conversion test * Fix crash in mc_amqp:size/1 Body can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0/ ct-system t=java ``` on branch native-amqp. * Fix crash in lists:flatten/1 Data can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp. * Fix crash in rabbit_writer Running test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp resulted in the following crash: ``` crasher: initial call: rabbit_writer:enter_mainloop/2 pid: <0.711.0> registered_name: [] exception error: bad argument in function size/1 called as size([<<0>>,<<"Sw">>,[<<160,2>>,<<"hi">>]]) *** argument 1: not tuple or binary in call from rabbit_binary_generator:build_content_frames/7 (rabbit_binary_generator.erl, line 89) in call from rabbit_binary_generator:build_simple_content_frames/4 (rabbit_binary_generator.erl, line 61) in call from rabbit_writer:assemble_frames/5 (rabbit_writer.erl, line 334) in call from rabbit_writer:internal_send_command_async/3 (rabbit_writer.erl, line 365) in call from rabbit_writer:handle_message/2 (rabbit_writer.erl, line 265) in call from rabbit_writer:handle_message/3 (rabbit_writer.erl, line 232) in call from rabbit_writer:mainloop1/2 (rabbit_writer.erl, line 223) ``` because #content.payload_fragments_rev is currently supposed to be a flat list of binaries instead of being an iolist. This commit fixes this crash inefficiently by calling iolist_to_binary/1. A better solution would be to allow AMQP legacy's #content.payload_fragments_rev to be an iolist. * Add accidentally deleted line back * mc: optimise mc_amqp internal format By removint the outer records for message and delivery annotations as well as application properties and footers. * mc: optimis mc_amqp map_add by using upsert * mc: refactoring and bug fixes * mc_SUITE routingheader assertions * mc remove serialize/1 callback as only used by amqp * mc_amqp: avoid returning a nested list from protocol_state * test and bug fix * move infer_type to mc_util * mc fixes and additiona assertions * Support headers exchange routing for MQTT messages When a headers exchange is bound to the MQTT topic exchange, routing will be performend based on both MQTT topic (by the topic exchange) and MQTT User Property (by the headers exchange). This combines the best worlds of both MQTT 5.0 and AMQP 0.9.1 and enables powerful routing topologies. When the User Property contains the same name multiple times, only the last name (and value) will be considered by the headers exchange. * Fix crash when sending from stream to amqpl When publishing a message via the stream protocol and consuming it via AMQP 0.9.1, the following crash occurred prior to this commit: ``` crasher: initial call: rabbit_channel:init/1 pid: <0.818.0> registered_name: [] exception exit: {{badmatch,undefined}, [{rabbit_channel,handle_deliver0,4, [{file,"rabbit_channel.erl"}, {line,2728}]}, {lists,foldl,3,[{file,"lists.erl"},{line,1594}]}, {rabbit_channel,handle_cast,2, [{file,"rabbit_channel.erl"}, {line,728}]}, {gen_server2,handle_msg,2, [{file,"gen_server2.erl"},{line,1056}]}, {proc_lib,wake_up,3, [{file,"proc_lib.erl"},{line,251}]}]} ``` This commit first gives `mc:init/3` the chance to set exchange and routing_keys annotations. If not set, `rabbit_stream_queue` will set these annotations assuming the message was originally published via the stream protocol. * Support consistent hash exchange routing for MQTT 5.0 When a consistent hash exchange is bound to the MQTT topic exchange, MQTT 5.0 messages can be routed to queues consistently based on the Correlation-Data in the PUBLISH packet. * Convert MQTT 5.0 User Property * to AMQP 0.9.1 headers * from AMQP 0.9.1 headers * to AMQP 1.0 application properties and message annotations * from AMQP 1.0 application properties and message annotations * Make use of Annotations in mc_mqtt:protocol_state/2 mc_mqtt:protocol_state/2 includes Annotations as parameter. It's cleaner to make use of these Annotations when computing the protocol state instead of relying on the caller (rabbitmq_mqtt_processor) to compute the protocol state. * Enforce AMQP 0.9.1 field name length limit The AMQP 0.9.1 spec prohibits field names longer than 128 characters. Therefore, when converting AMQP 1.0 message annotations, application properties or MQTT 5.0 User Property to AMQP 0.9.1 headers, drop any names longer than 128 characters. * Fix type specs Apply feedback from Michael Davis Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * Add mc_mqtt unit test suite Implement mc_mqtt:x_header/2 * Translate indicator that payload is UTF-8 encoded when converting between MQTT 5.0 and AMQP 1.0 * Translate single amqp-value section from AMQP 1.0 to MQTT Convert to a text representation, if possible, and indicate to MQTT client that the payload is UTF-8 encoded. This way, the MQTT client will be able to parse the payload. If conversion to text representation is not possible, encode the payload using the AMQP 1.0 type system and indiate the encoding via Content-Type message/vnd.rabbitmq.amqp. This Content-Type is not registered. Type "message" makes sense since it's a message. Vendor tree "vnd.rabbitmq.amqp" makes sense since merely subtype "amqp" is not registered. * Fix payload conversion * Translate Response Topic between MQTT and AMQP Translate MQTT 5.0 Response Topic to AMQP 1.0 reply-to address and vice versa. The Response Topic must be a UTF-8 encoded string. This commit re-uses the already defined RabbitMQ target addresses: ``` "/topic/" RK Publish to amq.topic with routing key RK "/exchange/" X "/" RK Publish to exchange X with routing key RK ``` By default, the MQTT topic exchange is configure dto be amq.topic using the 1st target address. When an operator modifies the mqtt.exchange, the 2nd target address is used. * Apply PR feedback and fix formatting Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * tidy up * Add MQTT message_containers test * consistent hash exchange: avoid amqp legacy conversion When hashing on a header value. * Avoid converting to amqp legacy when using exchange federation * Fix test flake * test and dialyzer fixes * dialyzer fix * Add MQTT protocol interoperability tests Test receiving from and sending to MQTT 5.0 and * AMQP 0.9.1 * AMQP 1.0 * STOMP * Streams * Regenerate portions of deps/rabbit/app.bzl with gazelle I'm not exactly sure how this happened, but gazell seems to have been run with an older version of the rules_erlang gazelle extension at some point. This caused generation of a structure that is no longer used. This commit updates the structure to the current pattern. * mc: refactoring * mc_amqpl: handle delivery annotations Just in case they are included. Also use iolist_to_iovec to create flat list of binaries when converting from amqp with amqp encoded payload. --------- Co-authored-by: David Ansari <david.ansari@gmx.de> Co-authored-by: Michael Davis <mcarsondavis@gmail.com> Co-authored-by: Rin Kuryloski <kuryloskip@vmware.com>
2023-08-31 18:27:13 +08:00
- mc_mqtt
- rabbit_mqtt
- rabbit_mqtt_confirms
- rabbit_mqtt_ff
- rabbit_mqtt_internal_event_handler
- rabbit_mqtt_keepalive
- rabbit_mqtt_packet
- rabbit_mqtt_processor
- rabbit_mqtt_qos0_queue
- rabbit_mqtt_reader
- rabbit_mqtt_retained_msg_store
- rabbit_mqtt_retained_msg_store_dets
- rabbit_mqtt_retained_msg_store_ets
- rabbit_mqtt_retained_msg_store_noop
- rabbit_mqtt_retainer
- rabbit_mqtt_retainer_sup
- rabbit_mqtt_sup
- rabbit_mqtt_util
rabbitmq_peer_discovery_aws:
- rabbit_peer_discovery_aws
- rabbitmq_peer_discovery_aws
rabbitmq_peer_discovery_common:
- rabbit_peer_discovery_cleanup
- rabbit_peer_discovery_common_app
- rabbit_peer_discovery_common_sup
- rabbit_peer_discovery_config
- rabbit_peer_discovery_httpc
- rabbit_peer_discovery_util
rabbitmq_peer_discovery_consul:
- rabbit_peer_discovery_consul
- rabbitmq_peer_discovery_consul
- rabbitmq_peer_discovery_consul_app
- rabbitmq_peer_discovery_consul_health_check_helper
- rabbitmq_peer_discovery_consul_sup
rabbitmq_peer_discovery_etcd:
- rabbit_peer_discovery_etcd
- rabbitmq_peer_discovery_etcd
- rabbitmq_peer_discovery_etcd_app
- rabbitmq_peer_discovery_etcd_sup
- rabbitmq_peer_discovery_etcd_v3_client
rabbitmq_peer_discovery_k8s:
- rabbit_peer_discovery_k8s
- rabbitmq_peer_discovery_k8s
- rabbitmq_peer_discovery_k8s_app
- rabbitmq_peer_discovery_k8s_node_monitor
- rabbitmq_peer_discovery_k8s_sup
rabbitmq_prelaunch:
- rabbit_boot_state
- rabbit_boot_state_sup
- rabbit_boot_state_systemd
- rabbit_boot_state_xterm_titlebar
- rabbit_logger_fmt_helpers
- rabbit_logger_json_fmt
- rabbit_logger_std_h
- rabbit_logger_text_fmt
- rabbit_prelaunch
- rabbit_prelaunch_app
- rabbit_prelaunch_conf
- rabbit_prelaunch_dist
- rabbit_prelaunch_early_logging
- rabbit_prelaunch_erlang_compat
- rabbit_prelaunch_errors
- rabbit_prelaunch_file
- rabbit_prelaunch_sighandler
- rabbit_prelaunch_sup
rabbitmq_prometheus:
- prometheus_process_collector
- prometheus_rabbitmq_alarm_metrics_collector
- prometheus_rabbitmq_core_metrics_collector
2023-09-20 12:03:13 +08:00
- prometheus_rabbitmq_dynamic_collector
- prometheus_rabbitmq_global_metrics_collector
- rabbit_prometheus_app
- rabbit_prometheus_dispatcher
- rabbit_prometheus_handler
rabbitmq_random_exchange:
- rabbit_exchange_type_random
rabbitmq_recent_history_exchange:
- rabbit_db_rh_exchange
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
- rabbit_db_rh_exchange_m2k_converter
- rabbit_exchange_type_recent_history
rabbitmq_sharding:
- rabbit_sharding_exchange_decorator
- rabbit_sharding_exchange_type_modulus_hash
- rabbit_sharding_interceptor
- rabbit_sharding_policy_validator
- rabbit_sharding_shard
- rabbit_sharding_util
rabbitmq_shovel:
- Elixir.RabbitMQ.CLI.Ctl.Commands.DeleteShovelCommand
- Elixir.RabbitMQ.CLI.Ctl.Commands.RestartShovelCommand
- Elixir.RabbitMQ.CLI.Ctl.Commands.ShovelStatusCommand
- rabbit_amqp091_shovel
- rabbit_amqp10_shovel
- rabbit_log_shovel
- rabbit_shovel
- rabbit_shovel_behaviour
- rabbit_shovel_config
- rabbit_shovel_dyn_worker_sup
- rabbit_shovel_dyn_worker_sup_sup
- rabbit_shovel_locks
- rabbit_shovel_parameters
- rabbit_shovel_status
- rabbit_shovel_sup
- rabbit_shovel_util
- rabbit_shovel_worker
- rabbit_shovel_worker_sup
rabbitmq_shovel_management:
2024-08-17 10:34:24 +08:00
- rabbit_shovel_mgmt_shovel
- rabbit_shovel_mgmt_shovels
- rabbit_shovel_mgmt_util
rabbitmq_shovel_prometheus:
- rabbit_shovel_prometheus_app
- rabbit_shovel_prometheus_collector
- rabbit_shovel_prometheus_sup
rabbitmq_stomp:
- Elixir.RabbitMQ.CLI.Ctl.Commands.ListStompConnectionsCommand
- rabbit_stomp
- rabbit_stomp_client_sup
- rabbit_stomp_connection_info
- rabbit_stomp_frame
- rabbit_stomp_internal_event_handler
- rabbit_stomp_processor
- rabbit_stomp_reader
- rabbit_stomp_sup
- rabbit_stomp_util
rabbitmq_stream:
- Elixir.RabbitMQ.CLI.Ctl.Commands.AddSuperStreamCommand
- Elixir.RabbitMQ.CLI.Ctl.Commands.DeleteSuperStreamCommand
- Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamConnectionsCommand
- Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamConsumerGroupsCommand
- Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamConsumersCommand
- Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamGroupConsumersCommand
- Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamPublishersCommand
2023-10-11 12:03:50 +08:00
- Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamTrackingCommand
- rabbit_stream
- rabbit_stream_connection_sup
- rabbit_stream_manager
- rabbit_stream_metrics
- rabbit_stream_metrics_gc
- rabbit_stream_reader
- rabbit_stream_sup
- rabbit_stream_utils
rabbitmq_stream_common:
- rabbit_stream_core
rabbitmq_stream_management:
- rabbit_stream_connection_consumers_mgmt
- rabbit_stream_connection_mgmt
- rabbit_stream_connection_publishers_mgmt
- rabbit_stream_connections_mgmt
- rabbit_stream_connections_vhost_mgmt
- rabbit_stream_consumers_mgmt
- rabbit_stream_management_utils
- rabbit_stream_mgmt_db
- rabbit_stream_publishers_mgmt
2023-10-11 12:03:50 +08:00
- rabbit_stream_tracking_mgmt
rabbitmq_top:
- rabbit_top_app
- rabbit_top_extension
- rabbit_top_sup
- rabbit_top_util
- rabbit_top_wm_ets_tables
- rabbit_top_wm_process
- rabbit_top_wm_processes
- rabbit_top_worker
rabbitmq_tracing:
- rabbit_tracing_app
- rabbit_tracing_consumer
- rabbit_tracing_consumer_sup
- rabbit_tracing_files
- rabbit_tracing_mgmt
- rabbit_tracing_sup
- rabbit_tracing_traces
- rabbit_tracing_util
- rabbit_tracing_wm_file
- rabbit_tracing_wm_files
- rabbit_tracing_wm_trace
- rabbit_tracing_wm_traces
rabbitmq_trust_store:
- rabbit_trust_store
- rabbit_trust_store_app
- rabbit_trust_store_certificate_provider
- rabbit_trust_store_file_provider
- rabbit_trust_store_http_provider
- rabbit_trust_store_sup
rabbitmq_web_dispatch:
- rabbit_cowboy_middleware
- rabbit_cowboy_redirect
- rabbit_cowboy_stream_h
- rabbit_web_dispatch
2023-06-24 05:52:56 +08:00
- rabbit_web_dispatch_access_control
- rabbit_web_dispatch_app
- rabbit_web_dispatch_listing_handler
- rabbit_web_dispatch_registry
- rabbit_web_dispatch_sup
- rabbit_web_dispatch_util
- webmachine_log
- webmachine_log_handler
rabbitmq_web_mqtt:
2024-03-17 01:38:11 +08:00
- Elixir.RabbitMQ.CLI.Ctl.Commands.ListWebMqttConnectionsCommand
- rabbit_web_mqtt_app
- rabbit_web_mqtt_handler
- rabbit_web_mqtt_stream_handler
rabbitmq_web_mqtt_examples:
- rabbit_web_mqtt_examples_app
rabbitmq_web_stomp:
- rabbit_web_stomp_app
- rabbit_web_stomp_connection_sup
- rabbit_web_stomp_handler
- rabbit_web_stomp_internal_event_handler
- rabbit_web_stomp_listener
- rabbit_web_stomp_middleware
- rabbit_web_stomp_stream_handler
- rabbit_web_stomp_sup
rabbitmq_web_stomp_examples:
- rabbit_web_stomp_examples_app
ranch:
- ranch
- ranch_acceptor
- ranch_acceptors_sup
- ranch_app
- ranch_conns_sup
- ranch_conns_sup_sup
- ranch_crc32c
- ranch_embedded_sup
- ranch_listener_sup
- ranch_protocol
- ranch_proxy_header
- ranch_server
- ranch_server_proxy
- ranch_ssl
- ranch_sup
- ranch_tcp
- ranch_transport
2024-05-23 05:42:29 +08:00
rebar3_format:
- default_formatter
- erlfmt_formatter
- otp_formatter
- rebar3_ast_formatter
- rebar3_format
- rebar3_format_prv
- rebar3_formatter
- sr_formatter
recon:
- recon
- recon_alloc
- recon_lib
- recon_map
- recon_rec
- recon_trace
redbug:
- redbug
- redbug_compiler
- redbug_dtop
- redbug_lexer
- redbug_parser
- redbug_targ
seshat:
- seshat
- seshat_app
- seshat_counters_server
- seshat_sup
stdout_formatter:
- stdout_formatter
- stdout_formatter_paragraph
- stdout_formatter_table
- stdout_formatter_utils
syslog:
- syslog
- syslog_error_h
- syslog_lager_backend
- syslog_lib
- syslog_logger
- syslog_logger_h
- syslog_monitor
- syslog_rfc3164
- syslog_rfc5424
sysmon_handler:
- sysmon_handler_app
- sysmon_handler_example_handler
- sysmon_handler_filter
- sysmon_handler_sup
- sysmon_handler_testhandler
systemd:
- systemd
- systemd_app
- systemd_journal_h
- systemd_kmsg_formatter
- systemd_protocol
- systemd_socket
- systemd_sup
- systemd_watchdog
thoas:
- thoas
- thoas_decode
- thoas_encode
trust_store_http:
- trust_store_http
- trust_store_http_app
- trust_store_http_sup
- trust_store_invalid_handler
- trust_store_list_handler