当前位置: 首页 > news >正文

Flink Rest Basic Auth - 安全认证

背景

公司目前需要将Flink实时作业云化,构建多租户实时计算平台。目前考虑为了资源高效利用,并不打算为每个租户部署一套独立的Kubernetes集群。也就意味着多个租户的作业可能会运行在同一套kubernets集群中。此时实时作业的任务就变的很危险,因为网络可能是通的,就会存在危险的REST API暴露出去,被一些不坏好意的人利用,从而影响其他租户的作业。鉴于此考虑给Flink的作业添加一个认证方式,可以是Kerberos或者是Http 用户名密码Baisc认证。各种搜索和询问,最终发现了一些线索FLIP-181: Custom netty HTTP request inbound/outbound handlers 这里描述了为何flink官方否定这个诉求。当然不要着急,笔者在flink-basic-auth-handler上找到了方案,并且成功将方案迁移到了flink-1.17.2版本中。

改造步骤

Flink 的JobManager/SQLGateway是基于Netty实现的一套轻量级的web服务接口,这些接口都实现了RestServerEndpoint抽象类。因此我们可以看看这个类start方法中可以看到在启动的代码中可以看到InboundChannelHandlerFactory这个东西,通过改Factory创建一个Inbound的hander。

public final void start() throws Exception {synchronized (lock) {Preconditions.checkState(state == State.CREATED, "The RestServerEndpoint cannot be restarted.");log.info("Starting rest endpoint.");final Router router = new Router();final CompletableFuture<String> restAddressFuture = new CompletableFuture<>();handlers = initializeHandlers(restAddressFuture);/* sort the handlers such that they are ordered the following:* /jobs* /jobs/overview* /jobs/:jobid* /jobs/:jobid/config* /:**/Collections.sort(handlers, RestHandlerUrlComparator.INSTANCE);checkAllEndpointsAndHandlersAreUnique(handlers);handlers.forEach(handler -> registerHandler(router, handler, log));ChannelInitializer<SocketChannel> initializer =new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws ConfigurationException {RouterHandler handler = new RouterHandler(router, responseHeaders);// SSL should be the first handler in the pipelineif (isHttpsEnabled()) {ch.pipeline().addLast("ssl",new RedirectingSslHandler(restAddress,restAddressFuture,sslHandlerFactory));}ch.pipeline().addLast(new HttpServerCodec()).addLast(new FileUploadHandler(uploadDir)).addLast(new FlinkHttpObjectAggregator(maxContentLength, responseHeaders));for (InboundChannelHandlerFactory factory :inboundChannelHandlerFactories) {Optional<ChannelHandler> channelHandler =factory.createHandler(configuration, responseHeaders);if (channelHandler.isPresent()) {ch.pipeline().addLast(channelHandler.get());}}ch.pipeline().addLast(new ChunkedWriteHandler()).addLast(handler.getName(), handler).addLast(new PipelineErrorHandler(log, responseHeaders));}};NioEventLoopGroup bossGroup =new NioEventLoopGroup(1, new ExecutorThreadFactory("flink-rest-server-netty-boss"));NioEventLoopGroup workerGroup =new NioEventLoopGroup(0, new ExecutorThreadFactory("flink-rest-server-netty-worker"));bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(initializer);Iterator<Integer> portsIterator;try {portsIterator = NetUtils.getPortRangeFromString(restBindPortRange);} catch (IllegalConfigurationException e) {throw e;} catch (Exception e) {throw new IllegalArgumentException("Invalid port range definition: " + restBindPortRange);}int chosenPort = 0;while (portsIterator.hasNext()) {try {chosenPort = portsIterator.next();final ChannelFuture channel;if (restBindAddress == null) {channel = bootstrap.bind(chosenPort);} else {channel = bootstrap.bind(restBindAddress, chosenPort);}serverChannel = channel.syncUninterruptibly().channel();break;} catch (final Exception e) {// syncUninterruptibly() throws checked exceptions via Unsafe// continue if the exception is due to the port being in use, fail early// otherwiseif (!(e instanceof java.net.BindException)) {throw e;}}}if (serverChannel == null) {throw new BindException("Could not start rest endpoint on any port in port range "+ restBindPortRange);}log.debug("Binding rest endpoint to {}:{}.", restBindAddress, chosenPort);final InetSocketAddress bindAddress = (InetSocketAddress) serverChannel.localAddress();final String advertisedAddress;if (bindAddress.getAddress().isAnyLocalAddress()) {advertisedAddress = this.restAddress;} else {advertisedAddress = bindAddress.getAddress().getHostAddress();}port = bindAddress.getPort();log.info("Rest endpoint listening at {}:{}", advertisedAddress, port);restBaseUrl = new URL(determineProtocol(), advertisedAddress, port, "").toString();restAddressFuture.complete(restBaseUrl);state = State.RUNNING;startInternal();}}

然后在构造函数中可以发现inboundChannelHandlerFactories对象是通过SPI方案加载进来的。

 public RestServerEndpoint(Configuration configuration)throws IOException, ConfigurationException {Preconditions.checkNotNull(configuration);RestServerEndpointConfiguration restConfiguration =RestServerEndpointConfiguration.fromConfiguration(configuration);Preconditions.checkNotNull(restConfiguration);this.configuration = configuration;this.restAddress = restConfiguration.getRestAddress();this.restBindAddress = restConfiguration.getRestBindAddress();this.restBindPortRange = restConfiguration.getRestBindPortRange();this.sslHandlerFactory = restConfiguration.getSslHandlerFactory();this.uploadDir = restConfiguration.getUploadDir();

相关文章:

Flink Rest Basic Auth - 安全认证

背景 公司目前需要将Flink实时作业云化,构建多租户实时计算平台。目前考虑为了资源高效利用,并不打算为每个租户部署一套独立的Kubernetes集群。也就意味着多个租户的作业可能会运行在同一套kubernets集群中。此时实时作业的任务就变的很危险,因为网络可能是通的,就会存在…...

安全U盘和普通U盘有什么区别?

安全U盘&#xff08;也称为加密U盘或安全闪存驱动器&#xff09;与普通U盘肯定是有一些区别的&#xff0c;从字面意思上来看&#xff0c;就能看出&#xff0c;安全U盘是能够保护文件数据安全性的&#xff0c;普通U盘没这一些功能的&#xff0c;可随意拷贝文件&#xff0c;不防盗…...

大数据与数据科学的学科边界

大数据和数据科学是两个紧密相关但又不完全相同的学科。它们都关注数据的收集、管理、分析和解释&#xff0c;但侧重点有所不同。 大数据主要关注处理和分析大规模数据集的技术和方法。它涉及到数据存储、数据处理、数据挖掘、数据可视化和分布式计算等方面的技术。大数据的目…...

Chrome 源码阅读:跟踪一个鼠标事件的流程

我们通过在关键节点打断点的方式&#xff0c;去分析一个鼠标事件的流程。 我们知道chromium是多进程模型&#xff0c;那么&#xff0c;我们可以推测&#xff1a;一个鼠标消息先从主进程产生&#xff0c;再通过跨进程通信发送给渲染进程&#xff0c;渲染进程再发送给WebFrame&a…...

[C/C++]_[初级]_[在Windows和macOS平台上导出动态库的一些思考]

场景 最近看了《COM本质论》里关于如何设计基于抽象基类作为二进制接口,把编译器和链接器的实现隐藏在这个二进制接口中,从而使用该DLL时不需要重新编译。在编译出C接口时,发现接口名直接是函数名,比如BindNativePort,怎么不是_BindNativePort?说明 VC++导出的函数默认是使…...

MySQL排序操作

025排序操作 select .. from .. order by 字段 asc/descselect empno, ename, sal from emp order by sal asc;asc 不写的话&#xff0c;默认升序 多个字段排序 查询员工的编号、姓名、薪资&#xff0c;按照薪资升序排列&#xff0c;如果薪资相同的&#xff0c;再按照姓名升…...

问题:西周后期形成了能够传布四方、留存后世的兵书——著述年代最早的兵书——( )和( ). #媒体#知识分享

问题&#xff1a;西周后期形成了能够传布四方、留存后世的兵书——著述年代最早的兵书——( )和( ). A、《军志》 B、《军事》 C、《军政》 D、《孙子兵法》 参考答案如图所示...

kafka-消费者-指定offset消费(SpringBoot整合Kafka)

文章目录 1、指定offset消费1.1、创建消费者监听器‘1.2、application.yml配置1.3、使用 Java代码 创建 主题 my_topic1 并建立3个分区并给每个分区建立3个副本1.4、创建生产者发送消息1.4.1、分区0中的数据 1.5、创建SpringBoot启动类1.6、屏蔽 kafka debug 日志 logback.xml1…...

JavaWeb2-Vue

Vue 前端框架&#xff0c;免除原生JS中的DOM操作简化书写 &#xff08;以前学过又忘了&#xff0c;现在才知道原来vue是前端的&#xff09; 基于MVVM思想&#xff08;model-view -viewModel&#xff09;实现数据双向绑定 model是数据模型 view负责数据展示 即DOM 中间这个负责…...

《广告数据定量分析》读书笔记之统计原理2

3.相关分析&#xff1a;描述的是两个数值变量间关系的强度。&#xff08;两个数值型变量之间的关系&#xff09; &#xff08;1&#xff09;图表表示&#xff1a;散点图 &#xff08;2&#xff09;衡量关系强度指标&#xff1a;相关系数r。 &#xff08;r的取值为-1到 1&…...

计算机视觉与模式识别实验2-2 SIFT特征提取与匹配

文章目录 &#x1f9e1;&#x1f9e1;实验流程&#x1f9e1;&#x1f9e1;SIFT算法原理总结&#xff1a;实现SIFT特征检测和匹配通过RANSAC 实现图片拼接更换其他图片再次测试效果&#xff08;依次进行SIFT特征提取、RANSAC 拼接&#xff09; &#x1f9e1;&#x1f9e1;全部代…...

kerberos: Clock skew too great (37) - PROCESS_TGS

kerberos认证失败错误信息&#xff1a; Caused by: org.ietf.jgss.GSSException: No valid credentials provided (Mechanism level: Clock skew too great (37) - PROCESS_TGS)at sun.security.jgss.krb5.Krb5Context.initSecContext(Krb5Context.java:772)at sun.security.j…...

【MATLAB高级编程】入门篇 | 向量化编程

【入门篇】向量化编程 1. 什么是向量?2. 向量的创建2.1 行向量2.2 列向量2.3 使用冒号运算符2.4 使用`linspace`和`logspace`3. 向量的基本操作3.1 向量元素访问3.2 向量的长度3.3 向量的加法和减法3.4 向量的点乘和叉乘3.5 向量的元素乘法和除法4. 向量的高级操作4.1 逻辑索引…...

Debezium日常分享系列之:Debezium 2.7.0.Beta1发布

Debezium日常分享系列之&#xff1a;Debezium 2.7.0.Beta1发布 一、重大变化1.快照工件2.Oracle 二、新功能和改进1.在 z/OS 上支持 Db22.NATS JetStream 接收器身份验证改进3.JDBC 接收器 MariaDB 方言支持4.JMX 导出器添加到 Debezium 服务器5.使用 Debezium Operator 启用 J…...

eNSP学习——RIP的水平分割和触发更新

目录 主要命令 原理概述 实验目的 实验内容 实验拓扑 实验编址 实验步骤 1、基本配置 2、搭建RIP网络 3、验证触发更新 4&#xff0e;验证水平分割 5、验证毒性逆转 需要eNSP各种配置命令的点击链接自取&#xff1a;华为&#xff45;NSP各种设备配置命令大全PDF版_…...

华为面经整理

文章目录 实习第一面准备提问相关算法相关 第一面结果提问环节 总结 实习 第一面准备 提问相关 操作系统有哪些功能 进程管理&#xff1a; 进程调度、进程同步和通信、多任务处理 内存管理&#xff1a; 内存分配、虚拟内存技术、内存保护 文件系统管理&#xff1a; 文件存储…...

数据恢复工具推荐:电脑回收站删除的文件怎么恢复?8个回收站恢复软件,收藏!

当文件从电脑的回收站被删除后&#xff0c;许多用户可能认为这些文件已永久丢失。然而&#xff0c;实际上&#xff0c;在数据被新数据覆盖之前&#xff0c;这些删除的文件仍然可以通过使用专门的数据恢复软件来恢复。本文将介绍8款顶级的文件恢复软件&#xff0c;恢复电脑回收站…...

Java 执行字符串 GroovyShell

1...

前端之npm运行时配置文件.npmrc(可用于配置npm淘宝源)

文章目录 前端之npm运行时配置文件.npmrc什么是.npmrc设置项目配置文件设置用户配置文件设置全局配置文件给npm 命令添加注册源选项 前端之npm运行时配置文件.npmrc 什么是.npmrc 官网&#xff1a;https://nodejs.cn/npm/cli/v7/configuring-npm/npmrc/ .npmrc&#xff0c;可…...

如何充分利用代理IP扩大网络接触面

目录 前言 第一部分&#xff1a;什么是代理IP&#xff1f; 第二部分&#xff1a;如何获取代理IP&#xff1f; 1. IP质量 2. 匿名性 3. 限制 第三部分&#xff1a;如何使用代理IP&#xff1f; 第四部分&#xff1a;如何充分利用代理IP&#xff1f; 总结&#xff1a; 前…...

变量 varablie 声明- Rust 变量 let mut 声明与 C/C++ 变量声明对比分析

一、变量声明设计&#xff1a;let 与 mut 的哲学解析 Rust 采用 let 声明变量并通过 mut 显式标记可变性&#xff0c;这种设计体现了语言的核心哲学。以下是深度解析&#xff1a; 1.1 设计理念剖析 安全优先原则&#xff1a;默认不可变强制开发者明确声明意图 let x 5; …...

<6>-MySQL表的增删查改

目录 一&#xff0c;create&#xff08;创建表&#xff09; 二&#xff0c;retrieve&#xff08;查询表&#xff09; 1&#xff0c;select列 2&#xff0c;where条件 三&#xff0c;update&#xff08;更新表&#xff09; 四&#xff0c;delete&#xff08;删除表&#xf…...

模型参数、模型存储精度、参数与显存

模型参数量衡量单位 M&#xff1a;百万&#xff08;Million&#xff09; B&#xff1a;十亿&#xff08;Billion&#xff09; 1 B 1000 M 1B 1000M 1B1000M 参数存储精度 模型参数是固定的&#xff0c;但是一个参数所表示多少字节不一定&#xff0c;需要看这个参数以什么…...

2025年能源电力系统与流体力学国际会议 (EPSFD 2025)

2025年能源电力系统与流体力学国际会议&#xff08;EPSFD 2025&#xff09;将于本年度在美丽的杭州盛大召开。作为全球能源、电力系统以及流体力学领域的顶级盛会&#xff0c;EPSFD 2025旨在为来自世界各地的科学家、工程师和研究人员提供一个展示最新研究成果、分享实践经验及…...

【Redis技术进阶之路】「原理分析系列开篇」分析客户端和服务端网络诵信交互实现(服务端执行命令请求的过程 - 初始化服务器)

服务端执行命令请求的过程 【专栏简介】【技术大纲】【专栏目标】【目标人群】1. Redis爱好者与社区成员2. 后端开发和系统架构师3. 计算机专业的本科生及研究生 初始化服务器1. 初始化服务器状态结构初始化RedisServer变量 2. 加载相关系统配置和用户配置参数定制化配置参数案…...

连锁超市冷库节能解决方案:如何实现超市降本增效

在连锁超市冷库运营中&#xff0c;高能耗、设备损耗快、人工管理低效等问题长期困扰企业。御控冷库节能解决方案通过智能控制化霜、按需化霜、实时监控、故障诊断、自动预警、远程控制开关六大核心技术&#xff0c;实现年省电费15%-60%&#xff0c;且不改动原有装备、安装快捷、…...

什么是库存周转?如何用进销存系统提高库存周转率?

你可能听说过这样一句话&#xff1a; “利润不是赚出来的&#xff0c;是管出来的。” 尤其是在制造业、批发零售、电商这类“货堆成山”的行业&#xff0c;很多企业看着销售不错&#xff0c;账上却没钱、利润也不见了&#xff0c;一翻库存才发现&#xff1a; 一堆卖不动的旧货…...

【项目实战】通过多模态+LangGraph实现PPT生成助手

PPT自动生成系统 基于LangGraph的PPT自动生成系统&#xff0c;可以将Markdown文档自动转换为PPT演示文稿。 功能特点 Markdown解析&#xff1a;自动解析Markdown文档结构PPT模板分析&#xff1a;分析PPT模板的布局和风格智能布局决策&#xff1a;匹配内容与合适的PPT布局自动…...

华为OD机试-食堂供餐-二分法

import java.util.Arrays; import java.util.Scanner;public class DemoTest3 {public static void main(String[] args) {Scanner in new Scanner(System.in);// 注意 hasNext 和 hasNextLine 的区别while (in.hasNextLine()) { // 注意 while 处理多个 caseint a in.nextIn…...

Python爬虫(二):爬虫完整流程

爬虫完整流程详解&#xff08;7大核心步骤实战技巧&#xff09; 一、爬虫完整工作流程 以下是爬虫开发的完整流程&#xff0c;我将结合具体技术点和实战经验展开说明&#xff1a; 1. 目标分析与前期准备 网站技术分析&#xff1a; 使用浏览器开发者工具&#xff08;F12&…...