base.c File Reference

Kafka global structures. More...

#include <freeradius-devel/kafka/base.h>
#include <freeradius-devel/server/cf_parse.h>
#include <freeradius-devel/server/tmpl.h>
#include <freeradius-devel/util/value.h>
#include <freeradius-devel/util/sbuff.h>
#include <freeradius-devel/util/size.h>

Include dependency graph for base.c:

Go to the source code of this file.

Data Structures
struct	fr_kafka_conf_ctx_t
struct	fr_kafka_conf_t
struct	fr_kafka_topic_conf_t

Macros
#define	BASE_CONFIG

Functions
static int	_kafka_conf_free (fr_kafka_conf_t *kc)
	Destroy a kafka configuration handle. More...
static int	_kafka_topic_conf_free (fr_kafka_topic_conf_t *ktc)
	Destroy a kafka topic configuration handle. More...
static fr_kafka_conf_t *	kafka_conf_from_cs (CONF_SECTION *cs)
static int	kafka_config_dflt (CONF_PAIR **out, void *parent, CONF_SECTION *cs, fr_token_t quote, conf_parser_t const *rule)
	Return the default value from the kafka client library. More...
static int	kafka_config_dflt_single (CONF_PAIR **out, UNUSED void *parent, CONF_SECTION *cs, char const *value, fr_token_t quote, conf_parser_t const *rule)
	Perform any conversions necessary to map kafka defaults to our values. More...
static int	kafka_config_parse (TALLOC_CTX *ctx, UNUSED void *out, UNUSED void *base, CONF_ITEM *ci, conf_parser_t const *rule)
	Translate config items directly to settings in a kafka config struct. More...
static int	kafka_config_parse_single (char const **out, CONF_PAIR *cp, conf_parser_t const *rule)
static fr_kafka_topic_conf_t *	kafka_topic_conf_from_cs (CONF_SECTION *cs)
static int	kafka_topic_config_dflt (CONF_PAIR **out, void *parent, CONF_SECTION *cs, fr_token_t quote, conf_parser_t const *rule)
	Return the default value for a topic from the kafka client library. More...
static int	kafka_topic_config_parse (UNUSED TALLOC_CTX *ctx, UNUSED void *out, UNUSED void *base, CONF_ITEM *ci, conf_parser_t const *rule)
	Translate config items directly to settings in a kafka topic config struct. More...

Variables
conf_parser_t const	kafka_base_consumer_config []
static conf_parser_t const	kafka_base_consumer_topic_config []
static conf_parser_t const	kafka_base_consumer_topics_config []
conf_parser_t const	kafka_base_producer_config []
static conf_parser_t const	kafka_base_producer_topic_config []
static conf_parser_t const	kafka_base_producer_topics_config []
static fr_table_ptr_sorted_t	kafka_check_cert_cn_table []
static size_t	kafka_check_cert_cn_table_len = NUM_ELEMENTS(kafka_check_cert_cn_table)
static conf_parser_t const	kafka_connection_config []
static conf_parser_t const	kafka_consumer_group_config []
static conf_parser_t const	kafka_metadata_config []
static conf_parser_t const	kafka_sasl_config []
static conf_parser_t const	kafka_sasl_kerberos_config []
static conf_parser_t const	kafka_sasl_oauth_config []
static conf_parser_t const	kafka_tls_config []
static conf_parser_t const	kafka_version_config []

Detailed Description

Kafka global structures.

Id

2604532ca9bcdeee528dc11559ec6eac7095f16e

Copyright

2022 Arran Cudbard-Bell (a.cud.nosp@m.bard.nosp@m.b@fre.nosp@m.erad.nosp@m.ius.o.nosp@m.rg)

Definition in file base.c.

---

Data Structure Documentation

fr_kafka_conf_ctx_t

struct fr_kafka_conf_ctx_t

Definition at line 40 of file base.c.

Collaboration diagram for fr_kafka_conf_ctx_t:

Data Fields
bool	empty_default	Don't produce messages saying the default is missing.
fr_table_ptr_sorted_t *	mapping	Mapping table between string constant.
size_t *	mapping_len	Length of the mapping tables.
char const *	property	Kafka configuration property.
size_t	size_scale	Divide/multiply FR_TYPE_SIZE by this amount.
char const *	string_sep	Used for multi-value configuration items. Kafka uses ', ' or ';' seemingly at random.

fr_kafka_conf_t

struct fr_kafka_conf_t

Definition at line 32 of file base.c.

Data Fields
rd_kafka_conf_t *	conf

fr_kafka_topic_conf_t

struct fr_kafka_topic_conf_t

Definition at line 36 of file base.c.

Data Fields
rd_kafka_topic_conf_t *	conf

Macro Definition Documentation

BASE_CONFIG

#define BASE_CONFIG

Value:

        { FR_CONF_FUNC("server", FR_TYPE_STRING, CONF_FLAG_REQUIRED | CONF_FLAG_MULTI, kafka_config_parse, kafka_config_dflt), \
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "metadata.broker.list", .string_sep = "," }}, \
        { FR_CONF_FUNC("client_id", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt), \
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "client.id" }}, \
        { FR_CONF_FUNC("rack_id", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt), \
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "client.rack" }}, \
        { FR_CONF_FUNC("request_max_size", FR_TYPE_SIZE, 0, kafka_config_parse, kafka_config_dflt), \
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "message.max.bytes" }}, \
        { FR_CONF_FUNC("request_copy_max_size", FR_TYPE_SIZE, 0, kafka_config_parse, kafka_config_dflt), \
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "message.copy.max.bytes" }}, \
        { FR_CONF_FUNC("response_max_size", FR_TYPE_SIZE, 0, kafka_config_parse, kafka_config_dflt), \
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "receive.message.max.bytes" }}, \
        { FR_CONF_FUNC("feature", FR_TYPE_STRING, CONF_FLAG_MULTI, kafka_config_parse, kafka_config_dflt), \
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "builtin.features", .string_sep = "," }}, \
        { FR_CONF_FUNC("debug", FR_TYPE_STRING, CONF_FLAG_MULTI, kafka_config_parse, kafka_config_dflt), \
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "debug", .string_sep = "," }}, \
        { FR_CONF_FUNC("plugin", FR_TYPE_STRING, CONF_FLAG_MULTI, kafka_config_parse, NULL), \
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "plugin.library.paths", .string_sep = ";" }}, \
        { FR_CONF_SUBSECTION_GLOBAL("metadata", 0, kafka_metadata_config) }, \
        { FR_CONF_SUBSECTION_GLOBAL("version", 0, kafka_version_config) }, \
        { FR_CONF_SUBSECTION_GLOBAL("connection", 0, kafka_connection_config) }, \
        { FR_CONF_SUBSECTION_GLOBAL("tls", 0, kafka_tls_config) }, \
        { FR_CONF_SUBSECTION_GLOBAL("sasl", 0, kafka_sasl_config) }

Definition at line 823 of file base.c.

Function Documentation

_kafka_conf_free()

static int _kafka_conf_free ( fr_kafka_conf_t *  kc )

static

Destroy a kafka configuration handle.

Parameters

[in]

kc

To destroy.

Returns

0

Definition at line 60 of file base.c.

Here is the caller graph for this function:

_kafka_topic_conf_free()

static int _kafka_topic_conf_free ( fr_kafka_topic_conf_t *  ktc )

static

Destroy a kafka topic configuration handle.

Parameters

[in]

ktc

To destroy.

Returns

0

Definition at line 90 of file base.c.

Here is the caller graph for this function:

kafka_conf_from_cs()

static fr_kafka_conf_t* kafka_conf_from_cs ( CONF_SECTION *  cs )

inlinestatic

Definition at line 67 of file base.c.

Here is the call graph for this function:

Here is the caller graph for this function:

kafka_config_dflt()

static int kafka_config_dflt ( CONF_PAIR **  out, void *  parent, CONF_SECTION *  cs, fr_token_t  quote, conf_parser_t const *  rule  )

static

Return the default value from the kafka client library.

Parameters

[out]	out	Where to write the pair.
[in]	parent	being populated.
[in]	cs	to allocate the pair in.
[in]	quote	to use when allocing the pair.
[in]	rule	UNUSED.

Returns

• 0 on success.
• -1 on failure.

Definition at line 204 of file base.c.

Here is the call graph for this function:

kafka_config_dflt_single()

static int kafka_config_dflt_single ( CONF_PAIR **  out, UNUSED void *  parent, CONF_SECTION *  cs, char const *  value, fr_token_t  quote, conf_parser_t const *  rule  )

static

Perform any conversions necessary to map kafka defaults to our values.

Parameters

[out]	out	Where to write the pair.
[in]	parent	being populated.
[in]	cs	to allocate the pair in.
[in]	value	to convert.
[in]	quote	to use when allocing the pair.
[in]	rule	UNUSED.

Returns

• 0 on success.
• -1 on failure.

Definition at line 127 of file base.c.

Here is the call graph for this function:

Here is the caller graph for this function:

kafka_config_parse()

static int kafka_config_parse ( TALLOC_CTX *  ctx, UNUSED void *  out, UNUSED void *  base, CONF_ITEM *  ci, conf_parser_t const *  rule  )

static

Translate config items directly to settings in a kafka config struct.

Parameters

[in]	ctx	to allocate fr_kafka_conf_t in.
[out]	out	Unused.
[in]	base	Unused.
[in]	ci	To parse.
[in]	rule	describing how to parse the item.

Returns

• 0 on success.
• -1 on failure

Definition at line 423 of file base.c.

Here is the call graph for this function:

kafka_config_parse_single()

static int kafka_config_parse_single ( char const **  out, CONF_PAIR *  cp, conf_parser_t const *  rule  )

static

Definition at line 324 of file base.c.

Here is the call graph for this function:

Here is the caller graph for this function:

kafka_topic_conf_from_cs()

static fr_kafka_topic_conf_t* kafka_topic_conf_from_cs ( CONF_SECTION *  cs )

inlinestatic

Definition at line 97 of file base.c.

Here is the call graph for this function:

Here is the caller graph for this function:

kafka_topic_config_dflt()

static int kafka_topic_config_dflt ( CONF_PAIR **  out, void *  parent, CONF_SECTION *  cs, fr_token_t  quote, conf_parser_t const *  rule  )

static

Return the default value for a topic from the kafka client library.

Parameters

[out]	out	Where to write the pair.
[in]	parent	being populated.
[in]	cs	to allocate the pair in.
[in]	quote	to use when allocing the pair.
[in]	rule	UNUSED.

Returns

• 0 on success.
• -1 on failure.

Definition at line 286 of file base.c.

Here is the call graph for this function:

kafka_topic_config_parse()

static int kafka_topic_config_parse ( UNUSED TALLOC_CTX *  ctx, UNUSED void *  out, UNUSED void *  base, CONF_ITEM *  ci, conf_parser_t const *  rule  )

static

Translate config items directly to settings in a kafka topic config struct.

Parameters

[in]	ctx	to allocate fr_kafka_conf_t in.
[out]	out	Unused.
[in]	base	Unused.
[in]	ci	To parse.
[in]	rule	describing how to parse the item.

Returns

• 0 on success.
• -1 on failure

Definition at line 495 of file base.c.

Here is the call graph for this function:

Variable Documentation

kafka_base_consumer_config

conf_parser_t const kafka_base_consumer_config[]

Definition at line 924 of file base.c.

kafka_base_consumer_topic_config

conf_parser_t const kafka_base_consumer_topic_config[]

static

Initial value:

= {
        
        { FR_CONF_FUNC("max_messages_per_cycle", FR_TYPE_UINT32, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "consume.callback.max.messages" }},
 
        
        { FR_CONF_FUNC("auto_offset_reset", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "auto.offset.reset" }},
 
        CONF_PARSER_TERMINATOR
}

Definition at line 889 of file base.c.

kafka_base_consumer_topics_config

conf_parser_t const kafka_base_consumer_topics_config[]

static

Initial value:

= {
        { FR_CONF_SUBSECTION_GLOBAL(CF_IDENT_ANY, CONF_FLAG_MULTI, kafka_base_consumer_topic_config) },
 
        CONF_PARSER_TERMINATOR
}

Definition at line 918 of file base.c.

kafka_base_producer_config

conf_parser_t const kafka_base_producer_config[]

Definition at line 1086 of file base.c.

kafka_base_producer_topic_config

conf_parser_t const kafka_base_producer_topic_config[]

static

Initial value:

= {
        
        { FR_CONF_FUNC("request_required_acks", FR_TYPE_INT16, 0, kafka_topic_config_parse, kafka_topic_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "request.required.acks" }},
 
        
        { FR_CONF_FUNC("request_timeout", FR_TYPE_TIME_DELTA, 0, kafka_topic_config_parse, kafka_topic_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "request.timeout.ms" }},
 
        
        { FR_CONF_FUNC("message_timeout", FR_TYPE_TIME_DELTA, 0, kafka_topic_config_parse, kafka_topic_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "message.timeout.ms" }},
 
        
        { FR_CONF_FUNC("partitioner", FR_TYPE_STRING, 0, kafka_topic_config_parse, kafka_topic_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "partitioner" }},
 
        
        { FR_CONF_FUNC("compression_type", FR_TYPE_STRING, 0, kafka_topic_config_parse, kafka_topic_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "compression.type" }},
 
        
        { FR_CONF_FUNC("compression_level", FR_TYPE_INT8, 0, kafka_topic_config_parse, kafka_topic_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "compression.level" }},
 
        CONF_PARSER_TERMINATOR
}

Definition at line 1029 of file base.c.

kafka_base_producer_topics_config

conf_parser_t const kafka_base_producer_topics_config[]

static

Initial value:

= {
        { FR_CONF_SUBSECTION_GLOBAL(CF_IDENT_ANY, 0, kafka_base_producer_topic_config) },
 
        CONF_PARSER_TERMINATOR
}

Definition at line 1080 of file base.c.

kafka_check_cert_cn_table

fr_table_ptr_sorted_t kafka_check_cert_cn_table[]

static

Initial value:

= {
        { L("false"),   "none"  },
        { L("no"),      "none"  },
        { L("true"),    "https" },
        { L("yes"),     "https" }
}

Definition at line 611 of file base.c.

kafka_check_cert_cn_table_len

size_t kafka_check_cert_cn_table_len = NUM_ELEMENTS(kafka_check_cert_cn_table)

static

Definition at line 617 of file base.c.

kafka_connection_config

conf_parser_t const kafka_connection_config[]

static

Definition at line 678 of file base.c.

kafka_consumer_group_config

conf_parser_t const kafka_consumer_group_config[]

static

Initial value:

= {
        
        { FR_CONF_FUNC("id", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "group.id" }},
 
        
        { FR_CONF_FUNC("instance_id", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "group.instance.id" }},
 
        
        { FR_CONF_FUNC("partition_assignment_strategy", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "partition.assignment.strategy" }},
 
        
        { FR_CONF_FUNC("session_timeout", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "session.timeout.ms" }},
 
        
        { FR_CONF_FUNC("heartbeat_interval", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "heartbeat.interval.ms" }},
 
        
        { FR_CONF_FUNC("coordinator_query_interval", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "coordinator.query.interval.ms" }},
 
 
        CONF_PARSER_TERMINATOR
}

Definition at line 848 of file base.c.

kafka_metadata_config

conf_parser_t const kafka_metadata_config[]

static

Initial value:

= {
        
        { FR_CONF_FUNC("refresh_interval", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "topic.metadata.refresh.interval.ms" }},
 
        
        { FR_CONF_FUNC("max_age", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "metadata.max.age.ms" }},
 
        
        { FR_CONF_FUNC("fast_refresh_interval", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "topic.metadata.refresh.fast.interval.ms" }},
 
        
        { FR_CONF_FUNC("max_propagation", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "topic.metadata.propagation.max.ms" }},
 
        
        { FR_CONF_FUNC("refresh_sparse", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "topic.metadata.refresh.sparse" }},
 
        
        { FR_CONF_FUNC("blacklist", FR_TYPE_STRING, CONF_FLAG_MULTI, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "topic.blacklist", .string_sep = ",", .empty_default = true }},
 
        CONF_PARSER_TERMINATOR
}

Definition at line 783 of file base.c.

kafka_sasl_config

conf_parser_t const kafka_sasl_config[]

static

Initial value:

= {
        
        { FR_CONF_FUNC("mech", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.mechanism" }},
 
        
        { FR_CONF_FUNC("username", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.username", .empty_default = true }},
 
        
        { FR_CONF_FUNC("password", FR_TYPE_STRING, CONF_FLAG_SECRET, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.password", .empty_default = true }},
 
        { FR_CONF_SUBSECTION_GLOBAL("kerberos", 0, kafka_sasl_kerberos_config) },
 
        { FR_CONF_SUBSECTION_GLOBAL("oauth", 0, kafka_sasl_oauth_config) },
 
        CONF_PARSER_TERMINATOR
}

Definition at line 585 of file base.c.

kafka_sasl_kerberos_config

conf_parser_t const kafka_sasl_kerberos_config[]

static

Initial value:

= {
        
        { FR_CONF_FUNC("service_name", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.kerberos.service.name" }},
 
        
        { FR_CONF_FUNC("principal", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.kerberos.principal" }},
 
        
        { FR_CONF_FUNC("kinit_cmd", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.kerberos.kinit.cmd" }},
 
        
        { FR_CONF_FUNC("keytab", FR_TYPE_STRING, CONF_FLAG_FILE_INPUT, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.kerberos.kinit.keytab", .empty_default = true }},
 
        
        { FR_CONF_FUNC("refresh_delay", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.kerberos.min.time.before.relogin" }},
 
        CONF_PARSER_TERMINATOR
}

Definition at line 551 of file base.c.

kafka_sasl_oauth_config

conf_parser_t const kafka_sasl_oauth_config[]

static

Initial value:

= {
        { FR_CONF_FUNC("oauthbearer_conf", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "sasl.oauthbearer.config", .empty_default = true }},
 
        { FR_CONF_FUNC("unsecure_jwt", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "enable.sasl.oauthbearer.unsecure.jwt" }},
 
        CONF_PARSER_TERMINATOR
}

Definition at line 541 of file base.c.

kafka_tls_config

conf_parser_t const kafka_tls_config[]

static

Definition at line 619 of file base.c.

kafka_version_config

conf_parser_t const kafka_version_config[]

static

Initial value:

= {
        
        { FR_CONF_FUNC("request", FR_TYPE_BOOL, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "api.version.request" }},
 
        
        { FR_CONF_FUNC("timeout", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "api.version.request.timeout.ms" }},
 
        
        { FR_CONF_FUNC("retry_delay", FR_TYPE_TIME_DELTA, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "api.version.fallback.ms" }},
 
        
        { FR_CONF_FUNC("default", FR_TYPE_STRING, 0, kafka_config_parse, kafka_config_dflt),
          .uctx = &(fr_kafka_conf_ctx_t){ .property = "broker.version.fallback" }},
 
        CONF_PARSER_TERMINATOR
}

Definition at line 755 of file base.c.