演练

    在深入研究代码之前，让我们先看看应用程序的体系架构，如 图7-1 所示。我们在这里构建三个重要的东西:事件源、服务和前端。 图中 Knative 内部的每一个组件都代表着我们将利用目前所学的知识来构建的内容，包括使用 Kaniko 构建模板的服务和用于轮询数据的自定义事件源:

    USGS 事件源

    我们将构建一个自定义的 ContainerSource 事件源，它将在给定的时间间隔轮询 USGS 提供的数据。为预构建的容器镜像打包。

    图 7-1 应用程序的体系结构。来自于 USGS 的地震数据源作为事件进入我们的事件源，这将触发我们的 GeoCoder 服务来持久化事件。我们的前台也将使用我们的 Geocoder 服务来查询最近的事件。

    Geocoder 服务

    这将为事件源提供 POST 事件的节点，并使用提供的坐标查找地址。它还将作为前端用来查询和检索最近的事件的节点。我们将使用 Build 服务来构建容器镜像，与运行在 Kubernetes 上的 Postgres 数据库通信。

    前端

    一个可以可视化最近的地震活动的轻量级的、持续运行的前端。

    我们可以使用 在 Kubernetes 集群上轻松地搭建起 Postgres 数据库，Helm 是一个可以轻松地在 Kubernetes 上打包和共享应用程序包的工具。关于如何在您的 Kubernetes 集群上启动和运行的介绍，请务必参考 Helm 的文档。如果您运行在 Minikube 或没有任何特定的权限要求的 Kubernetes 集群上，那么您可以使用以下简单的命令来设置 Helm:

    对于像谷歌的 GCP 这样具有更深层安全配置的集群，请参考 Helm Quickstart 指南。接下来我们可以设置一个 Postgres 数据库并且传递一些配置参数以使设置更容易：

    这将在我们的 Kubernetes 集群中创建一个 Postgres 数据库，将用户密码设置为 devPass ，并创建一个名为 geocode 的数据库。我们已经将 Postgres 服务器命名为 geocodedb ，这意味着在 Kubernetes 集群中，我们可以通过 geocodedb-postgresql.default.svc.cluster.local 访问该服务器。现在让我们来深入了解代码吧!

    Geocoder 服务

    如应用程序体系结构图所示，我们的事件源和前端都将向 Geocoder 服务发送请求，后者将与 Postgres 数据库通信。这将我们的服务置于应用程序的中心位置。对我们服务的 HTTP POST 请求将会在数据库中记录事件，而 GET 请求将检索过去24小时内发生的事件。让我们来看一下 示例 7-1 中我们服务的代码。

    示例 7-1 geocoder/app.rb

    1. require 'geocoder'
    2. require 'json'
    3. require 'pg'
    4. require 'sinatra'
    6. set :bind, '0.0.0.0'
    8. # DB connection credentials are passed via environment
    9. # variables
    10. DB_HOST = ENV["DB_HOST"] || 'localhost'
    11. DB_DATABASE = ENV["DB_DATABASE"] || 'geocode'
    12. DB_USER = ENV["DB_USER"] || 'postgres'
    13. DB_PASS = ENV["DB_PASS"] || 'password'
    15. # Connect to the database and create table if it doesn't exist
    16. conn = PG.connect( dbname: DB_DATABASE, host: DB_HOST, 
    17.     password: DB_PASS, user: DB_USER)
    18. conn.exec "CREATE TABLE IF NOT EXISTS events (
    19.     id varchar(20) NOT NULL PRIMARY KEY,
    20.     timestamp timestamp,
    21.     lat double precision,
    22.     lon double precision,
    23.     mag real,
    24.     address text
    25. );"
    27. # Store an event
    28. post '/' do
    29.     d = JSON.parse(request.body.read.to_s)
    30.     address = coords_to_address(d["lat"], d["long"])
    31.     id = d["id"]
    33.     conn.prepare("insert_#{id}", 
    34.         'INSERT INTO events VALUES ($1, $2, $3, $4, $5, $6)')
    35.     conn.exec_prepared("insert_#{id}", [d["id"], d["time"], 
    36.         d["lat"], d["long"], d["mag"], address.to_json])
    37. end
    39. # Get all events from the last 24 hours
    40. get '/' do
    41.     select_statement = "select * from events where 
    42.         timestamp > 'now'::timestamp - '24 hours'::interval;"
    43.     results = conn.exec(select_statement)
    44.     jResults = []
    45.     results.each do |row|
    46.         jResults << row
    47.     end
    49.     content_type 'application/json'
    50.     headers 'Access-Control-Allow-Origin' => "*"
    51.     return jResults.to_json
    52. end
    54. # Get the address from a given set of coordinates
    55. def coords_to_address(lat, lon)
    56.     coords = [lat, lon]
    57.     results = Geocoder.search(coords)
    59.     a = results.first
    60.     address = {
    61.         address: a.address,
    62.         house_number: a.house_number,
    63.         street: a.street,
    64.         county: a.county,
    65.         city: a.city,
    66.         state: a.state,
    67.         state_code: a.state_code,
    68.         postal_code: a.postal_code,
    69.         country: a.country,
    70.         country_code: a.country_code,
    71.         coordinates: a.coordinates
    72.     }
    74. end

    我们将使用 Knative 为我们构建容器镜像，将连接到 Postgres 数据库所需的信息传递给它，并运行我们的服务。我们可以在 中看到这是如何设置的。

    示例 7-2. earthquake-demo/geocoder-service.yaml

    1. apiVersion: serving.knative.dev/v1alpha1
    2. kind: Service
    3. metadata:
    4.   name: geocoder
    6. spec:
    7.   runLatest:
    8.     configuration:
    9.       build:
    10.         serviceAccountName: build-bot
    11.         source:
    12.           git:
    13.            url: https://github.com/gswk/geocoder.git
    14.            revision: master
    15.         template:
    16.           name: kaniko
    17.           arguments:
    18.           - name: IMAGE
    19.             value: docker.io/gswk/geocoder
    20.       revisionTemplate:
    21.         spec:
    22.           container:
    23.             image: docker.io/gswk/geocoder
    24.             env:
    25.             - name: DB_HOST
    26.               value: "geocodedb-postgresql.default.svc.cluster.local"
    27.             - name: DB_DATABASE
    28.               value: "geocode"
    29.             - name: DB_USER
    30.               value: "postgres"
    31.             - name: DB_PASS
    32.               value: "devPass"

    kubectl apply -f earthquake-demo/geocoder-service.yaml

    我们的事件源将负责在指定的时间间隔内轮询 USGS 地震活动的数据，解析它，并将其发送到我们定义的接收器。由于我们需要轮询数据，并且没有由 USGS 将其推送给我们的可能，因此它非常适合使用 ContainerSource 编写自定义事件源。

    在设置事件源之前，还需要一个事件发送的通道。虽然我们可以直接将事件从事件源发送到我们的服务，但如果我们希望将来能够将事件发送到另一个服务，这将给我们带来一些灵活性。我们只需要一个简单的通道，我们将在 示例 7-3 中定义它。

    示例 7-3. earthquake-demo/channel.yaml

    kubectl apply -f earthquake-demo/channel.yaml

    正如我们在第6章中构建自定义事件源一样，我们的这个事件源也是由一个脚本构成，在本例中是一个 ruby 脚本，它接受两个命令行标志位: —sink 和 —interval。让我们在 中看看它。

    示例 7-4. usgs-event-source/usgs-event-source.rb

    1. require 'date'
    2. require "httparty"
    3. require 'json'
    4. require 'logger'
    5. require 'optimist'
    7. $stdout.sync = true
    8. @logger = Logger.new(STDOUT)
    9. @logger.level = Logger::DEBUG
    11. # Poll the USGS feed for real-time earthquake readings
    12. def pull_hourly_earthquake(lastTime, sink)
    13.     # Get all detected earthquakes in the last hour
    14.     url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/" \
    15.     + "summary/all_hour.geojson"
    16.     response = HTTParty.get(url)
    17.     j = JSON.parse(response.body)
    19.     # Keep track of latest recorded event, reporting all
    20.     # if none have been tracked so far
    21.     cycleLastTime = lastTime
    23.     # Parse each reading and emit new ones as events
    24.     j["features"].each do |f|
    25.         time = f["properties"]["time"]
    27.         if time > lastTime
    28.             msg = {
    29.                 time: DateTime.strptime(time.to_s,'%Q'),
    30.                 id: f["id"],
    31.                 mag: f["properties"]["mag"],
    32.                 lat: f["geometry"]["coordinates"][1],
    33.                 long: f["geometry"]["coordinates"][0]
    34.             }
    36.             publish_event(msg, sink)
    37.         end
    39.         # Keep track of latest reading
    40.         if time > cycleLastTime
    41.             cycleLastTime = time
    42.         end
    43.     end
    45.     lastTime = cycleLastTime
    46.     return lastTime
    48. # POST event to provided sink
    49. def publish_event(message, sink)
    51.     puts message.to_json
    52.     r = HTTParty.post(sink, 
    53.         :headers => {'Content-Type'=>'text/plain'}, 
    54.         :body => message.to_json)
    56.     if r.code != 200
    57.         @logger.error("Error! #{r}")
    58.     end
    59. end
    62. # Parse CLI flags
    63. opts = Optimist::options do
    64.     banner <<-EOS
    65. Poll USGS Real-Time Earthquake data
    66. Usage:
    67.   ruby usgs-event-source.rb
    68. EOS
    69.     opt :interval, "Poll Frequenvy", 
    70.         :default => 10
    71.     opt :sink, "Sink to send events", 
    72.         :default => "http://localhost:8080"
    73. end
    75. # Begin polling USGS data
    76. lastTime = 0
    77. @logger.info("Polling every #{opts[:interval]} seconds")
    78. while true do 
    79.     @logger.debug("Polling . . .")
    80.     lastTime = pull_hourly_earthquake(lastTime, opts[:sink])
    81.     sleep(opts[:interval])
    82. end

    像前面一样，Knative 在作为 ContainerSource 事件源运行时将处理 —sink 标志位。我们还提供了一个额外的标记 —interval，我们将定义这个标记，因为我们编写的代码将允许用户定义自己的轮询间隔。脚本被打包为 Docker 镜像并上传到 Dockerhub 上的 gswk/usgs-event-source 下。剩下的就是创建 中所示的事件源的 yaml，并创建订阅，以便将事件从通道发送到 示例 7-6 中所示的服务。

    示例 7-5. earthquake-demo/usgs-event-source.yaml

    1. apiVersion: sources.eventing.knative.dev/v1alpha1
    2. kind: ContainerSource
    3. metadata:
    4.   labels:
    5.     controller-tools.k8s.io: "1.0"
    6.   name: usgs-event-source
    7. spec:
    8.   image: docker.io/gswk/usgs-event-source:latest
    9.   args: 
    10.     - "--interval=10"
    11.   sink:
    12.     apiVersion: serving.knative.dev/v1alpha1
    13.     kind: Service
    14. name: geocoder

    $ kubectl apply -f earthquake-demo/subscription.yaml

    一旦我们应用这个 yaml，事件源将启动一个持续运行的容器，该容器将轮询事件并将它们发送到我们创建的通道中。另外，我们需要将 Geocoder 服务连接到通道中。

    示例 7-6. earthquake-demo/subscription.yaml

    $ kubectl apply -f earthquake-demo/subscription.yaml

    创建了订阅之后，我们已经将所有内容连接起来，以便将事件通过自定义事件源带到环境中，然后将它们发送到服务中，服务将把它们持久化到 Postgres 数据库中。我们还有最后一个要部署的部分，那就是我们的前端，用来可视化所有东西。

    前端

    最后，我们需要把我们收集的所有数据一起放在前端来进行可视化。我们创建了一个简单的网站，并将其打包在一个容器镜像中，该容器镜像将使用 Nginx 提供服务。当页面加载时，它将调用 Geocoder 服务，返回一个地震事件的数组，包括坐标和震级，并在地图上显示它们。我们还将把它设置为 Knative 服务，这样我们就可以免费获得简易的路由和度量。同样，我们将像其他 Knative 服务一样编写一个 yaml，并使用 Kaniko 构建模板，如 所示。

    示例 7-7. earthquake-demo/frontend/frontend-service.yaml

    1. apiVersion: serving.knative.dev/v1alpha1
    2. kind: Service
    3. metadata:
    4.   name: earthquake-demo
    5.   namespace: default
    6. spec:
    7.   runLatest:
    8.     configuration:
    9.       build:
    10.         serviceAccountName: build-bot
    11.         source:
    12.           git:
    13.             url: https://github.com/gswk/earthquake-demo-frontend.git
    14.             revision: master
    15.         template:
    16.           name: kaniko
    17.           arguments:
    18.           - name: IMAGE
    19.             value: docker.io/gswk/earthquake-demo-frontend
    20.       revisionTemplate:
    21.         spec:
    22.           container:
    23.             image: docker.io/gswk/earthquake-demo-frontend
    24.             env:
    25.             - name: EVENTS_API
    26. value: "http://geocoder.default.svc.cluster.local"

    $ kubectl apply -f earthquake-demo/frontend-service.yaml

    前端界面 图 7-2 我们的应用程序启动起来了

    当请求进入我们的前端应用程序时，它将从 Geocoder 服务中提取事件，当新事件进入时，它们将被我们的自定义事件源接收。此外，Knative 还提供了一些额外的工具，通过内置的日志记录、度量和跟踪功能，帮助您保持应用程序和服务的正常运行。

    任何在生产环境中运行过代码的人都知道我们的故事还没有结束。因为编写了代码和部署了应用程序，我们就需要对管理和运维负责。正确地了解代码如何处理日志及度量是该运维流程的一部分，幸运的是 Knative 附带了许多工具来提供这些信息。更好的是，它的大部分功能已经自动绑定到您的代码中，而不需要您做任何特殊的事情。

    让我们从深入研究 Geocoder 服务的日志开始，这个功能由 Kibana 提供，Kibana 是我们设置 Knative 的服务组件时安装的。在我们访问任何东西之前，我们需要在我们的 Kubernetes 集群中设置一个代理，只需一个命令就可以轻松完成:

    $ kubectl proxy

    这将为访问整个 Kubernetes 集群中打开一个代理，并可以在我们机器的8001端口上访问它。这也包括 Kibana，我们可以通过 http://localhost:8001/api/v1/namespaces/knative-monitoring/services/kibana-logging/proxy/app/kibana 访问它。

    我们需要提供一个索引模板，我们可以简单地使用 * 和 timestamp_millis 的时间过滤器。最后，如果我们转到 Kibana 的 Discover 选项卡，我们将看到系统中的每个日志！让我们看一下通过如下搜索方式发送到 Geocoder 服务的请求及其结果，如 所示。

    localEndpoint.serviceName = geocoder

    图 7-3 展示我们的Geocoder服务日志的Kibana仪表板

    那么，如果只想看粗略的度量标准呢？看看某些指标比如失败的请求和响应时间可以提供我们解决应用程序问题的线索，Knative 还通过与 Grafana 一起提供非常多的度量指标（从响应代码的分布到我们的服务使用了多少 CPU）来帮助我们解决这个问题。Knative 甚至包括一个仪表盘，用于可视化当前集群的使用情况，以帮助进行容量规划。在加载 Grafana 之前，我们需要使用以下命令将端口转发到 Kubernetes 集群:

    1. $ kubectl port-forward
    2.     --namespace knative-monitoring $(kubectl get pods
    3.     --namespace knative-monitoring
    4.     --output=jsonpath="{.items..metadata.name}") 3000

    一旦转发，我们可以通过 http://localhost:3000 访问仪表板。在 中，我们可以看到发送到 Geocoder 服务的请求的图，看起来很好很健康!

    Geocoder 图 7-4 对Geocoder服务的成功和失败请求对比的图表

    最后，Knative 还附带了 Zipkin 来帮助跟踪我们的请求。当请求通过我们的 ingress 网关进入，并到达数据库时，通过一些简单的仪表化，我们可以很好地了解我们的应用程序内部情况。在按照前述设置好代理之后，我们可以通过 http://localhost:8001/api/v1/namespaces/istio-system/services/zipkin:9411/proxy/Zipkin 来访问 Zipkin。一旦进入，我们就可以通过它看到请求如何发送到我们的 Geocoder服务上的，如 和 图 7-6 所示。

    图7-5 对一个到Geocoder服务请求的简单跟踪

    Geocoder_zipkin2 图 7-6 我们的服务请求堆栈时间分解

    结论

    一个完整的带有我们自己定制的事件源的应用程序成功了！这在很大程度上总结了我们在本书中要学习的内容，但是 Knative 还可以提供更多。同时，Knative 也在不断地发展和完善。当您继续您的学习旅程时，还有很多资源值得关注，所以在我们结束之前，我们需要知道我们在第8章中还提供其他一些参考资料。