Knative 入門系列7

Knative 是一個基于 Kubernetes 的,用于構建、部署和管理現代 serverless 應用的平臺。Getting Started with Knative 是一本由 Pivotal 公司贊助 O’Reilly 出品的電子書,公眾號后臺回復“knative”獲取英文版下載地址。本書中文版由 ServiceMesher 社區自發翻譯系列文章,這是該系列的第7章。

讓我們把我們所學的一切運用起來去創造一些東西吧!我們進行一個演練,它利用了您前面所學到的許多知識,并通過使用美國地質勘探局 (USGS) 地震數據源的數據提供了一個服務,以可視化地展示世界各地的地震活動。您可以在 GitHub 存儲庫 gswk/earthquakedemo 中找到我們將要介紹的代碼。

架構

在深入研究代碼之前,讓我們先看看應用程序的體系架構,如 圖7-1 所示。我們在這里構建三個重要的東西:事件源、服務和前端。圖中 Knative 內部的每一個組件都代表著我們將利用目前所學的知識來構建的內容,包括使用 Kaniko 構建模板的服務和用于輪詢數據的自定義事件源:

USGS 事件源我們將構建一個自定義的 ContainerSource 事件源,它將在給定的時間間隔輪詢 USGS 提供的數據。為預構建的容器鏡像打包。

arch圖 7-1 應用程序的體系結構。來自于 USGS 的地震數據源作為事件進入我們的事件源,這將觸發我們的 GeoCoder 服務來持久化事件。我們的前臺也將使用我們的 Geocoder 服務來查詢最近的事件。

Geocoder 服務這將為事件源提供 POST 事件的節點,并使用提供的坐標查找地址。它還將作為前端用來查詢和檢索最近的事件的節點。我們將使用 Build 服務來構建容器鏡像。與運行在 Kubernetes 上的 Postgres 數據庫通信。

前端一個可以可視化最近的地震活動的輕量級的、持續運行的前端

我們可以使用 Helm 在 Kubernetes 集群上輕松地搭建起 Postgres 數據庫,Helm 是一個可以輕松地在 Kubernetes 上打包和共享應用程序包的工具。關于如何在你的 Kubernetes 集群上啟動和運行的介紹,請務必參考 Helm 的文檔。如果您運行在 Minikube 或沒有任何特定的權限要求的 Kubernetes 集群上,那么您可以使用以下簡單的命令來設置 Helm: $ helm init

對于像谷歌的 GCP 這樣具有更深層安全配置的集群,請參考 Helm Quickstart 指南。接下來我們可以設置一個 Postgres 數據庫并且傳遞一些配置參數以使設置更容易:

$ helm install--name geocodedb--set postgresqlPassword=devPass,postgresqlDatabase                        =geocode stable/postgresql

這將在我們的 Kubernetes 集群中創建一個 Postgres 數據庫,將用戶密碼設置為 devPass ,并創建一個名為 geocode 的數據庫。我們已經將 Postgres 服務器命名為 geocodedb ,這意味著在 Kubernetes 集群中,我們可以通過 geocodedb-postgresql.default.svc.cluster.local 訪問該服務器?，F在讓我們來深入了解代碼吧!

Geocoder 服務

如應用程序體系結構圖所示,我們的事件源和前端都將向 Geocoder 服務發送請求,后者將與 Postgres 數據庫通信。這將我們的服務置于應用程序的中心位置。對我們服務的 HTTP POST 請求將會在數據庫中記錄事件,而 GET 請求將檢索過去24小時內發生的事件。讓我們來看一下 示例 7-1 中我們服務的代碼。

示例 7-1 geocoder/app.rb

require 'geocoder'require 'json'require 'pg'require 'sinatra'set :bind, '0.0.0.0'# DB connection credentials are passed via environment# variablesDB_HOST = ENV["DB_HOST"] || 'localhost'DB_DATABASE = ENV["DB_DATABASE"] || 'geocode'DB_USER = ENV["DB_USER"] || 'postgres'DB_PASS = ENV["DB_PASS"] || 'password'# Connect to the database and create table if it doesn't existconn = PG.connect( dbname: DB_DATABASE, host: DB_HOST,     password: DB_PASS, user: DB_USER)conn.exec "CREATE TABLE IF NOT EXISTS events (    id varchar(20) NOT NULL PRIMARY KEY,    timestamp timestamp,    lat double precision,    lon double precision,    mag real,    address text);"# Store an eventpost '/' do    d = JSON.parse(request.body.read.to_s)    address = coords_to_address(d["lat"], d["long"])    id = d["id"]    conn.prepare("insert_#{id}",         'insert INTO events VALUES ($1, $2, $3, $4, $5, $6)')    conn.exec_prepared("insert_#{id}", [d["id"], d["time"],         d["lat"], d["long"], d["mag"], address.to_json])end# Get all events from the last 24 hoursget '/' do    select_statement = "select * from events where         timestamp > 'now'::timestamp - '24 hours'::interval;"    results = conn.exec(select_statement)    jResults = []    results.each do |row|        jResults << row    end    content_type 'application/json'    headers 'Access-Control-Allow-Origin' => "*"    return jResults.to_jsonend# Get the address from a given set of coordinatesdef coords_to_address(lat, lon)    coords = [lat, lon]    results = Geocoder.search(coords)    a = results.first    address = {        address: a.address,        house_number: a.house_number,        street: a.street,        county: a.county,        city: a.city,        state: a.state,        state_code: a.state_code,        postal_code: a.postal_code,        country: a.country,        country_code: a.country_code,        coordinates: a.coordinates    }    return addressend

我們將使用 Knative 為我們構建容器鏡像,將連接到 Postgres 數據庫所需的信息傳遞給它,并運行我們的服務。我們可以在 示例7-2 中看到這是如何設置的。

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

apiVersion: serving.knative.dev/v1alpha1kind: Servicemetadata:  name: geocoder  namespace: defaultspec:  runLatest:    configuration:      build:        serviceAccountName: build-bot        source:          git:           url: https://github.com/gswk/geocoder.git           revision: master        template:          name: kaniko          arguments:          - name: IMAGE            value: docker.io/gswk/geocoder      revisionTemplate:        spec:          container:            image: docker.io/gswk/geocoder            env:            - name: DB_HOST              value: "geocodedb-postgresql.default.svc.cluster.local"            - name: DB_DATABASE              value: "geocode"            - name: DB_USER              value: "postgres"            - name: DB_PASS              value: "devPass"

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

因為我們已經通過環境變量傳遞了所有連接信息給我們的 Postgres 數據庫,這是我們的服務運行需要的所有信息。接下來,我們將獲取事件源并運行它,以便我們可以開始向新部署的服務發送事件。

USGS 事件源

我們的事件源將負責在指定的時間間隔內輪詢 USGS 地震活動的數據,解析它,并將其發送到我們定義的接收器。由于我們需要輪詢數據,并且沒有由 USGS 將其推送給我們的可能,因此它非常適合使用 ContainerSource 編寫自定義事件源。

在設置事件源之前,還需要一個事件發送的通道。雖然我們可以直接將事件從事件源發送到我們的服務,但如果我們希望將來能夠將事件發送到另一個服務,這將給我們帶來一些靈活性。我們只需要一個簡單的通道,我們將在 示例 7-3 中定義它。

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

apiVersion: eventing.knative.dev/v1alpha1kind: Channelmetadata:  name: geocode-channelspec:  provisioner:    apiVersion: eventing.knative.dev/v1alpha1    kind: ClusterChannelProvisionername: in-memory-channel

kubectl apply -f earthquake-demo/channel.yaml

正如我們在第6章中構建自定義事件源一樣,我們的這個事件源也是由一個腳本構成,在本例中是一個 Ruby 腳本,它接受兩個命令行標志位: --sink 和 --interval。讓我們在 示例7-4中看看這個。

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

require 'date'require "httparty"require 'json'require 'logger'require 'optimist'$stdout.sync = true@logger = Logger.new(STDOUT)@logger.level = Logger::DEBUG# Poll the USGS feed for real-time earthquake readingsdef pull_hourly_earthquake(lastTime, sink)    # Get all detected earthquakes in the last hour    url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/"     + "summary/all_hour.geojson"    response = HTTParty.get(url)    j = JSON.parse(response.body)    # Keep track of latest recorded event, reporting all    # if none have been tracked so far    cycleLastTime = lastTime    # Parse each reading and emit new ones as events    j["features"].each do |f|        time = f["properties"]["time"]        if time > lastTime            msg = {                time: DateTime.strptime(time.to_s,'%Q'),                id: f["id"],                mag: f["properties"]["mag"],                lat: f["geometry"]["coordinates"][1],                long: f["geometry"]["coordinates"][0]            }            publish_event(msg, sink)        end        # Keep track of latest reading        if time > cycleLastTime            cycleLastTime = time        end    end    lastTime = cycleLastTime    return lastTimeend# POST event to provided sinkdef publish_event(message, sink)    @logger.info("Sending #{message[:id]} to #{sink}")    puts message.to_json    r = HTTParty.post(sink,         :headers => {'Content-Type'=>'text/plain'},         :body => message.to_json)    if r.code != 200        @logger.error("Error! #{r}")    endend# Parse CLI flagsopts = Optimist::options do    banner <<-EOSPoll USGS Real-Time Earthquake dataUsage:  ruby usgs-event-source.rbEOS    opt :interval, "Poll Frequenvy",         :default => 10    opt :sink, "Sink to send events",         :default => "http://localhost:8080"end# Begin polling USGS datalastTime = 0@logger.info("Polling every #{opts[:interval]} seconds")while true do     @logger.debug("Polling . . .")    lastTime = pull_hourly_earthquake(lastTime, opts[:sink])    sleep(opts[:interval])end

像往常一樣,Knative 在作為 ContainerSource 事件源運行時將處理 --sink 標志位。我們還提供了一個額外的標記 --interval,我們將定義這個標記,因為我們編寫的代碼將允許用戶定義自己的輪詢間隔。腳本被打包為 Docker 鏡像并上傳到 Dockerhub 上的 gswk/usgs-event-source 下。剩下的就是創建 示例 7-5 中所示的我們的事件源的 YAML,并創建訂閱,以便將事件從通道發送到 示例 7-6 中所示的服務。

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

apiVersion: sources.eventing.knative.dev/v1alpha1kind: ContainerSourcemetadata:  labels:    controller-tools.k8s.io: "1.0"  name: usgs-event-sourcespec:  image: docker.io/gswk/usgs-event-source:latest  args:     - "--interval=10"  sink:    apiVersion: serving.knative.dev/v1alpha1    kind: Servicename: geocoder

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

一旦我們應用這個 YAML,事件源將啟動一個持續運行的容器,該容器將輪詢事件并將它們發送到我們創建的通道中。另外,我們需要將 Geocoder 服務連接到通道中。

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

apiVersion: eventing.knative.dev/v1alpha1kind: Subscriptionmetadata:  name: geocode-subscriptionspec:  channel:    apiVersion: eventing.knative.dev/v1alpha1    kind: Channel    name: geocode-channel  subscriber:    ref:      apiVersion: serving.knative.dev/v1alpha1      kind: Servicename: geocoder

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

創建了訂閱之后,我們已經將所有內容連接起來,以便將事件通過自定義事件源帶到環境中,然后將它們發送到服務中,服務將把它們持久化到 Postgres 數據庫中。我們還有最后一個要部署的部分,那就是我們的前端,用來可視化所有東西。

前端

最后,我們需要把我們收集的所有數據一起放在前端來進行可視化。我們創建了一個簡單的網站,并將其打包在一個容器鏡像中,該容器鏡像將使用 Nginx 提供服務。當頁面加載時,它將調用 Geocoder 服務,返回一個地震事件的數組,包括坐標和震級,并在地圖上顯示它們。我們還將把它設置為 Knative 服務,這樣我們就可以免費獲得簡易的路由和度量。同樣,我們將像其他 Knative 服務一樣編寫一個 YAML,并使用 Kaniko 構建模板,如 示例 7-7 所示。

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

apiVersion: serving.knative.dev/v1alpha1kind: Servicemetadata:  name: earthquake-demo  namespace: defaultspec:  runLatest:    configuration:      build:        serviceAccountName: build-bot        source:          git:            url: https://github.com/gswk/earthquake-demo-frontend.git            revision: master        template:          name: kaniko          arguments:          - name: IMAGE            value: docker.io/gswk/earthquake-demo-frontend      revisionTemplate:        spec:          container:            image: docker.io/gswk/earthquake-demo-frontend            env:            - name: EVENTS_APIvalue: "http://geocoder.default.svc.cluster.local"

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

我們定義 EVENTS_API 環境變量,前端將使用該變量來了解 Geocoder 服務的位置。最后這一部分就緒后,我們就可以啟動并運行整個系統了!我們的應用程序如 圖 7-2 所示。

前端界面圖 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 服務的請求及其結果,如 圖7-3 所示。

localEndpoint.serviceName = geocoder

Geocoder圖 7-3。展示我們的Geocoder服務日志的Kibana儀表板

那么,如果只想看粗略的度量標準呢?看看某些指標比如失敗的請求和響應時間可以提供解決我們應用程序問題的線索,Knative 還通過與 Grafana 一起提供非常多的度量指標(從響應代碼的分布到我們的服務使用了多少 CPU)來幫助我們解決這個問題。Knative 甚至包括一個儀表盤,用于可視化當前集群的使用情況,以幫助進行容量規劃。在加載 Grafana 之前,我們需要使用以下命令將端口轉發到 Kubernetes 集群:

$ kubectl port-forward    --namespace knative-monitoring $(kubectl get pods    --namespace knative-monitoring    --selector=app=grafana    --output=jsonpath="{.items..metadata.name}") 3000

一旦轉發,我們可以通過 http://localhost:3000 訪問儀表板。在 圖7-4 中,我們可以看到發送到 Geocoder 服務的請求的圖,看起來很好很健康!

Geocoder圖 7-4 對Geocoder服務的成功和失敗請求對比的圖表

最后,Knative 還附帶了 Zipkin 來幫助跟蹤我們的請求。當請求通過我們的 ingress 網關進入,并到達數據庫時,通過一些簡單的儀表化,我們可以很好地了解我們的應用程序內部情況。在按照前述設置好代理之后,我們可以通過 http://localhost:8001/api/v1/namespaces/istio-system/services/zipkin:9411/proxy/Zipkin 來訪問 Zipkin。一旦進入,我們就可以通過它看到請求如何發送到我們的 Geocoder服務上的,如 圖 7-5 和 圖 7-6 所示。

Geocoder_zipkin1圖7-5 對一個到Geocoder服務請求的簡單跟蹤

Geocoder_zipkin2圖 7-6 我們的服務請求堆棧時間分解

結論

成功了!一個完整的應用程序,帶有我們自己定制的事件源。這在很大程度上總結了我們在本書中要學習的內容,但是 Knative 還可以提供更多。同時,Knative 也在不斷地發展和完善。當你繼續你的旅程時,還有很多資源值得關注,所以在我們結束之前,我們需要知道我們在第8章中還提供其他一些參考資料。